import { exec, execSync } from 'child_process'; import fs from 'fs'; import path from 'path'; import makeWASocket, { DisconnectReason, WASocket, makeCacheableSignalKeyStore, useMultiFileAuthState, } from '@whiskeysockets/baileys'; import { CronExpressionParser } from 'cron-parser'; import { ASSISTANT_NAME, DATA_DIR, IPC_POLL_INTERVAL, MAIN_GROUP_FOLDER, POLL_INTERVAL, STORE_DIR, TIMEZONE, TRIGGER_PATTERN, } from './config.js'; import { AvailableGroup, runContainerAgent, writeGroupsSnapshot, writeTasksSnapshot, } from './container-runner.js'; import { createTask, deleteTask, getAllChats, getAllRegisteredGroups, getAllSessions, getAllTasks, getLastGroupSync, getMessagesSince, getNewMessages, getRouterState, getTaskById, initDatabase, setLastGroupSync, setRegisteredGroup, setRouterState, setSession, storeChatMetadata, storeMessage, updateChatName, updateTask, } from './db.js'; import { GroupQueue } from './group-queue.js'; import { startSchedulerLoop } from './task-scheduler.js'; import { RegisteredGroup } from './types.js'; import { logger } from './logger.js'; const GROUP_SYNC_INTERVAL_MS = 24 * 60 * 60 * 1000; // 24 hours let sock: WASocket; let lastTimestamp = ''; let sessions: Record = {}; let registeredGroups: Record = {}; let lastAgentTimestamp: Record = {}; // LID to phone number mapping (WhatsApp now sends LID JIDs for self-chats) let lidToPhoneMap: Record = {}; // Guards to prevent duplicate loops on WhatsApp reconnect let messageLoopRunning = false; let ipcWatcherRunning = false; let groupSyncTimerStarted = false; const queue = new GroupQueue(); /** * Translate a JID from LID format to phone format if we have a mapping. * Returns the original JID if no mapping exists. */ function translateJid(jid: string): string { if (!jid.endsWith('@lid')) return jid; const lidUser = jid.split('@')[0].split(':')[0]; const phoneJid = lidToPhoneMap[lidUser]; if (phoneJid) { logger.debug({ lidJid: jid, phoneJid }, 'Translated LID to phone JID'); return phoneJid; } return jid; } async function setTyping(jid: string, isTyping: boolean): Promise { try { await sock.sendPresenceUpdate(isTyping ? 'composing' : 'paused', jid); } catch (err) { logger.debug({ jid, err }, 'Failed to update typing status'); } } function loadState(): void { // Load from SQLite (migration from JSON happens in initDatabase) lastTimestamp = getRouterState('last_timestamp') || ''; const agentTs = getRouterState('last_agent_timestamp'); try { lastAgentTimestamp = agentTs ? JSON.parse(agentTs) : {}; } catch { logger.warn('Corrupted last_agent_timestamp in DB, resetting'); lastAgentTimestamp = {}; } sessions = getAllSessions(); registeredGroups = getAllRegisteredGroups(); logger.info( { groupCount: Object.keys(registeredGroups).length }, 'State loaded', ); } function saveState(): void { setRouterState('last_timestamp', lastTimestamp); setRouterState( 'last_agent_timestamp', JSON.stringify(lastAgentTimestamp), ); } function registerGroup(jid: string, group: RegisteredGroup): void { registeredGroups[jid] = group; setRegisteredGroup(jid, group); // Create group folder const groupDir = path.join(DATA_DIR, '..', 'groups', group.folder); fs.mkdirSync(path.join(groupDir, 'logs'), { recursive: true }); logger.info( { jid, name: group.name, folder: group.folder }, 'Group registered', ); } /** * Sync group metadata from WhatsApp. * Fetches all participating groups and stores their names in the database. * Called on startup, daily, and on-demand via IPC. */ async function syncGroupMetadata(force = false): Promise { // Check if we need to sync (skip if synced recently, unless forced) if (!force) { const lastSync = getLastGroupSync(); if (lastSync) { const lastSyncTime = new Date(lastSync).getTime(); const now = Date.now(); if (now - lastSyncTime < GROUP_SYNC_INTERVAL_MS) { logger.debug({ lastSync }, 'Skipping group sync - synced recently'); return; } } } try { logger.info('Syncing group metadata from WhatsApp...'); const groups = await sock.groupFetchAllParticipating(); let count = 0; for (const [jid, metadata] of Object.entries(groups)) { if (metadata.subject) { updateChatName(jid, metadata.subject); count++; } } setLastGroupSync(); logger.info({ count }, 'Group metadata synced'); } catch (err) { logger.error({ err }, 'Failed to sync group metadata'); } } /** * Get available groups list for the agent. * Returns groups ordered by most recent activity. */ function getAvailableGroups(): AvailableGroup[] { const chats = getAllChats(); const registeredJids = new Set(Object.keys(registeredGroups)); return chats .filter((c) => c.jid !== '__group_sync__' && c.jid.endsWith('@g.us')) .map((c) => ({ jid: c.jid, name: c.name, lastActivity: c.last_message_time, isRegistered: registeredJids.has(c.jid), })); } /** * Process all pending messages for a group. * Called by the GroupQueue when it's this group's turn. */ async function processGroupMessages(chatJid: string): Promise { const group = registeredGroups[chatJid]; if (!group) return true; const isMainGroup = group.folder === MAIN_GROUP_FOLDER; // Get all messages since last agent interaction const sinceTimestamp = lastAgentTimestamp[chatJid] || ''; const missedMessages = getMessagesSince( chatJid, sinceTimestamp, ASSISTANT_NAME, ); if (missedMessages.length === 0) return true; // For non-main groups, check if any message has the trigger if (!isMainGroup) { const hasTrigger = missedMessages.some((m) => TRIGGER_PATTERN.test(m.content.trim()), ); if (!hasTrigger) return true; } const lines = missedMessages.map((m) => { const escapeXml = (s: string) => s .replace(/&/g, '&') .replace(//g, '>') .replace(/"/g, '"'); return `${escapeXml(m.content)}`; }); const prompt = `\n${lines.join('\n')}\n`; logger.info( { group: group.name, messageCount: missedMessages.length }, 'Processing messages', ); await setTyping(chatJid, true); const response = await runAgent(group, prompt, chatJid); await setTyping(chatJid, false); if (response) { // Fix batching bug: advance to latest message in batch, not just the trigger lastAgentTimestamp[chatJid] = missedMessages[missedMessages.length - 1].timestamp; saveState(); await sendMessage(chatJid, `${ASSISTANT_NAME}: ${response}`); return true; } return false; } async function runAgent( group: RegisteredGroup, prompt: string, chatJid: string, ): Promise { const isMain = group.folder === MAIN_GROUP_FOLDER; const sessionId = sessions[group.folder]; // Update tasks snapshot for container to read (filtered by group) const tasks = getAllTasks(); writeTasksSnapshot( group.folder, isMain, tasks.map((t) => ({ id: t.id, groupFolder: t.group_folder, prompt: t.prompt, schedule_type: t.schedule_type, schedule_value: t.schedule_value, status: t.status, next_run: t.next_run, })), ); // Update available groups snapshot (main group only can see all groups) const availableGroups = getAvailableGroups(); writeGroupsSnapshot( group.folder, isMain, availableGroups, new Set(Object.keys(registeredGroups)), ); try { const output = await runContainerAgent( group, { prompt, sessionId, groupFolder: group.folder, chatJid, isMain, }, (proc, containerName) => queue.registerProcess(chatJid, proc, containerName), ); if (output.newSessionId) { sessions[group.folder] = output.newSessionId; setSession(group.folder, output.newSessionId); } if (output.status === 'error') { logger.error( { group: group.name, error: output.error }, 'Container agent error', ); return null; } return output.result; } catch (err) { logger.error({ group: group.name, err }, 'Agent error'); return null; } } async function sendMessage(jid: string, text: string): Promise { try { await sock.sendMessage(jid, { text }); logger.info({ jid, length: text.length }, 'Message sent'); } catch (err) { logger.error({ jid, err }, 'Failed to send message'); } } function startIpcWatcher(): void { if (ipcWatcherRunning) { logger.debug('IPC watcher already running, skipping duplicate start'); return; } ipcWatcherRunning = true; const ipcBaseDir = path.join(DATA_DIR, 'ipc'); fs.mkdirSync(ipcBaseDir, { recursive: true }); const processIpcFiles = async () => { // Scan all group IPC directories (identity determined by directory) let groupFolders: string[]; try { groupFolders = fs.readdirSync(ipcBaseDir).filter((f) => { const stat = fs.statSync(path.join(ipcBaseDir, f)); return stat.isDirectory() && f !== 'errors'; }); } catch (err) { logger.error({ err }, 'Error reading IPC base directory'); setTimeout(processIpcFiles, IPC_POLL_INTERVAL); return; } for (const sourceGroup of groupFolders) { const isMain = sourceGroup === MAIN_GROUP_FOLDER; const messagesDir = path.join(ipcBaseDir, sourceGroup, 'messages'); const tasksDir = path.join(ipcBaseDir, sourceGroup, 'tasks'); // Process messages from this group's IPC directory try { if (fs.existsSync(messagesDir)) { const messageFiles = fs .readdirSync(messagesDir) .filter((f) => f.endsWith('.json')); for (const file of messageFiles) { const filePath = path.join(messagesDir, file); try { const data = JSON.parse(fs.readFileSync(filePath, 'utf-8')); if (data.type === 'message' && data.chatJid && data.text) { // Authorization: verify this group can send to this chatJid const targetGroup = registeredGroups[data.chatJid]; if ( isMain || (targetGroup && targetGroup.folder === sourceGroup) ) { await sendMessage( data.chatJid, `${ASSISTANT_NAME}: ${data.text}`, ); logger.info( { chatJid: data.chatJid, sourceGroup }, 'IPC message sent', ); } else { logger.warn( { chatJid: data.chatJid, sourceGroup }, 'Unauthorized IPC message attempt blocked', ); } } fs.unlinkSync(filePath); } catch (err) { logger.error( { file, sourceGroup, err }, 'Error processing IPC message', ); const errorDir = path.join(ipcBaseDir, 'errors'); fs.mkdirSync(errorDir, { recursive: true }); fs.renameSync( filePath, path.join(errorDir, `${sourceGroup}-${file}`), ); } } } } catch (err) { logger.error( { err, sourceGroup }, 'Error reading IPC messages directory', ); } // Process tasks from this group's IPC directory try { if (fs.existsSync(tasksDir)) { const taskFiles = fs .readdirSync(tasksDir) .filter((f) => f.endsWith('.json')); for (const file of taskFiles) { const filePath = path.join(tasksDir, file); try { const data = JSON.parse(fs.readFileSync(filePath, 'utf-8')); // Pass source group identity to processTaskIpc for authorization await processTaskIpc(data, sourceGroup, isMain); fs.unlinkSync(filePath); } catch (err) { logger.error( { file, sourceGroup, err }, 'Error processing IPC task', ); const errorDir = path.join(ipcBaseDir, 'errors'); fs.mkdirSync(errorDir, { recursive: true }); fs.renameSync( filePath, path.join(errorDir, `${sourceGroup}-${file}`), ); } } } } catch (err) { logger.error({ err, sourceGroup }, 'Error reading IPC tasks directory'); } } setTimeout(processIpcFiles, IPC_POLL_INTERVAL); }; processIpcFiles(); logger.info('IPC watcher started (per-group namespaces)'); } async function processTaskIpc( data: { type: string; taskId?: string; prompt?: string; schedule_type?: string; schedule_value?: string; context_mode?: string; groupFolder?: string; chatJid?: string; // For register_group jid?: string; name?: string; folder?: string; trigger?: string; containerConfig?: RegisteredGroup['containerConfig']; }, sourceGroup: string, // Verified identity from IPC directory isMain: boolean, // Verified from directory path ): Promise { switch (data.type) { case 'schedule_task': if ( data.prompt && data.schedule_type && data.schedule_value && data.groupFolder ) { // Authorization: non-main groups can only schedule for themselves const targetGroup = data.groupFolder; if (!isMain && targetGroup !== sourceGroup) { logger.warn( { sourceGroup, targetGroup }, 'Unauthorized schedule_task attempt blocked', ); break; } // Resolve the correct JID for the target group (don't trust IPC payload) const targetJid = Object.entries(registeredGroups).find( ([, group]) => group.folder === targetGroup, )?.[0]; if (!targetJid) { logger.warn( { targetGroup }, 'Cannot schedule task: target group not registered', ); break; } const scheduleType = data.schedule_type as 'cron' | 'interval' | 'once'; let nextRun: string | null = null; if (scheduleType === 'cron') { try { const interval = CronExpressionParser.parse(data.schedule_value, { tz: TIMEZONE, }); nextRun = interval.next().toISOString(); } catch { logger.warn( { scheduleValue: data.schedule_value }, 'Invalid cron expression', ); break; } } else if (scheduleType === 'interval') { const ms = parseInt(data.schedule_value, 10); if (isNaN(ms) || ms <= 0) { logger.warn( { scheduleValue: data.schedule_value }, 'Invalid interval', ); break; } nextRun = new Date(Date.now() + ms).toISOString(); } else if (scheduleType === 'once') { const scheduled = new Date(data.schedule_value); if (isNaN(scheduled.getTime())) { logger.warn( { scheduleValue: data.schedule_value }, 'Invalid timestamp', ); break; } nextRun = scheduled.toISOString(); } const taskId = `task-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; const contextMode = data.context_mode === 'group' || data.context_mode === 'isolated' ? data.context_mode : 'isolated'; createTask({ id: taskId, group_folder: targetGroup, chat_jid: targetJid, prompt: data.prompt, schedule_type: scheduleType, schedule_value: data.schedule_value, context_mode: contextMode, next_run: nextRun, status: 'active', created_at: new Date().toISOString(), }); logger.info( { taskId, sourceGroup, targetGroup, contextMode }, 'Task created via IPC', ); } break; case 'pause_task': if (data.taskId) { const task = getTaskById(data.taskId); if (task && (isMain || task.group_folder === sourceGroup)) { updateTask(data.taskId, { status: 'paused' }); logger.info( { taskId: data.taskId, sourceGroup }, 'Task paused via IPC', ); } else { logger.warn( { taskId: data.taskId, sourceGroup }, 'Unauthorized task pause attempt', ); } } break; case 'resume_task': if (data.taskId) { const task = getTaskById(data.taskId); if (task && (isMain || task.group_folder === sourceGroup)) { updateTask(data.taskId, { status: 'active' }); logger.info( { taskId: data.taskId, sourceGroup }, 'Task resumed via IPC', ); } else { logger.warn( { taskId: data.taskId, sourceGroup }, 'Unauthorized task resume attempt', ); } } break; case 'cancel_task': if (data.taskId) { const task = getTaskById(data.taskId); if (task && (isMain || task.group_folder === sourceGroup)) { deleteTask(data.taskId); logger.info( { taskId: data.taskId, sourceGroup }, 'Task cancelled via IPC', ); } else { logger.warn( { taskId: data.taskId, sourceGroup }, 'Unauthorized task cancel attempt', ); } } break; case 'refresh_groups': // Only main group can request a refresh if (isMain) { logger.info( { sourceGroup }, 'Group metadata refresh requested via IPC', ); await syncGroupMetadata(true); // Write updated snapshot immediately const availableGroups = getAvailableGroups(); writeGroupsSnapshot( sourceGroup, true, availableGroups, new Set(Object.keys(registeredGroups)), ); } else { logger.warn( { sourceGroup }, 'Unauthorized refresh_groups attempt blocked', ); } break; case 'register_group': // Only main group can register new groups if (!isMain) { logger.warn( { sourceGroup }, 'Unauthorized register_group attempt blocked', ); break; } if (data.jid && data.name && data.folder && data.trigger) { registerGroup(data.jid, { name: data.name, folder: data.folder, trigger: data.trigger, added_at: new Date().toISOString(), containerConfig: data.containerConfig, }); } else { logger.warn( { data }, 'Invalid register_group request - missing required fields', ); } break; default: logger.warn({ type: data.type }, 'Unknown IPC task type'); } } async function connectWhatsApp(): Promise { const authDir = path.join(STORE_DIR, 'auth'); fs.mkdirSync(authDir, { recursive: true }); const { state, saveCreds } = await useMultiFileAuthState(authDir); sock = makeWASocket({ auth: { creds: state.creds, keys: makeCacheableSignalKeyStore(state.keys, logger), }, printQRInTerminal: false, logger, browser: ['NanoClaw', 'Chrome', '1.0.0'], }); sock.ev.on('connection.update', (update) => { const { connection, lastDisconnect, qr } = update; if (qr) { const msg = 'WhatsApp authentication required. Run /setup in Claude Code.'; logger.error(msg); exec( `osascript -e 'display notification "${msg}" with title "NanoClaw" sound name "Basso"'`, ); setTimeout(() => process.exit(1), 1000); } if (connection === 'close') { const reason = (lastDisconnect?.error as any)?.output?.statusCode; const shouldReconnect = reason !== DisconnectReason.loggedOut; logger.info({ reason, shouldReconnect }, 'Connection closed'); if (shouldReconnect) { logger.info('Reconnecting...'); connectWhatsApp(); } else { logger.info('Logged out. Run /setup to re-authenticate.'); process.exit(0); } } else if (connection === 'open') { logger.info('Connected to WhatsApp'); // Build LID to phone mapping from auth state for self-chat translation if (sock.user) { const phoneUser = sock.user.id.split(':')[0]; const lidUser = sock.user.lid?.split(':')[0]; if (lidUser && phoneUser) { lidToPhoneMap[lidUser] = `${phoneUser}@s.whatsapp.net`; logger.debug({ lidUser, phoneUser }, 'LID to phone mapping set'); } } // Sync group metadata on startup (respects 24h cache) syncGroupMetadata().catch((err) => logger.error({ err }, 'Initial group sync failed'), ); // Set up daily sync timer (only once) if (!groupSyncTimerStarted) { groupSyncTimerStarted = true; setInterval(() => { syncGroupMetadata().catch((err) => logger.error({ err }, 'Periodic group sync failed'), ); }, GROUP_SYNC_INTERVAL_MS); } startSchedulerLoop({ sendMessage, registeredGroups: () => registeredGroups, getSessions: () => sessions, queue, onProcess: (groupJid, proc, containerName) => queue.registerProcess(groupJid, proc, containerName), }); startIpcWatcher(); startMessageLoop(); recoverPendingMessages(); } }); sock.ev.on('creds.update', saveCreds); sock.ev.on('messages.upsert', ({ messages }) => { for (const msg of messages) { if (!msg.message) continue; const rawJid = msg.key.remoteJid; if (!rawJid || rawJid === 'status@broadcast') continue; // Translate LID JID to phone JID if applicable const chatJid = translateJid(rawJid); const timestamp = new Date( Number(msg.messageTimestamp) * 1000, ).toISOString(); // Always store chat metadata for group discovery storeChatMetadata(chatJid, timestamp); // Only store full message content for registered groups if (registeredGroups[chatJid]) { storeMessage( msg, chatJid, msg.key.fromMe || false, msg.pushName || undefined, ); } } }); } async function startMessageLoop(): Promise { if (messageLoopRunning) { logger.debug('Message loop already running, skipping duplicate start'); return; } messageLoopRunning = true; // Wire up the queue's message processing function queue.setProcessMessagesFn(processGroupMessages); logger.info(`NanoClaw running (trigger: @${ASSISTANT_NAME})`); while (true) { try { const jids = Object.keys(registeredGroups); const { messages, newTimestamp } = getNewMessages( jids, lastTimestamp, ASSISTANT_NAME, ); if (messages.length > 0) { logger.info({ count: messages.length }, 'New messages'); // Advance the "seen" cursor for all messages immediately lastTimestamp = newTimestamp; saveState(); // Deduplicate by group and enqueue const groupsWithMessages = new Set(); for (const msg of messages) { groupsWithMessages.add(msg.chat_jid); } for (const chatJid of groupsWithMessages) { queue.enqueueMessageCheck(chatJid); } } } catch (err) { logger.error({ err }, 'Error in message loop'); } await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL)); } } /** * Startup recovery: check for unprocessed messages in registered groups. * Handles crash between advancing lastTimestamp and processing messages. */ function recoverPendingMessages(): void { for (const [chatJid, group] of Object.entries(registeredGroups)) { const sinceTimestamp = lastAgentTimestamp[chatJid] || ''; const pending = getMessagesSince(chatJid, sinceTimestamp, ASSISTANT_NAME); if (pending.length > 0) { logger.info( { group: group.name, pendingCount: pending.length }, 'Recovery: found unprocessed messages', ); queue.enqueueMessageCheck(chatJid); } } } function ensureContainerSystemRunning(): void { try { execSync('container system status', { stdio: 'pipe' }); logger.debug('Apple Container system already running'); } catch { logger.info('Starting Apple Container system...'); try { execSync('container system start', { stdio: 'pipe', timeout: 30000 }); logger.info('Apple Container system started'); } catch (err) { logger.error({ err }, 'Failed to start Apple Container system'); console.error( '\n╔════════════════════════════════════════════════════════════════╗', ); console.error( '║ FATAL: Apple Container system failed to start ║', ); console.error( '║ ║', ); console.error( '║ Agents cannot run without Apple Container. To fix: ║', ); console.error( '║ 1. Install from: https://github.com/apple/container/releases ║', ); console.error( '║ 2. Run: container system start ║', ); console.error( '║ 3. Restart NanoClaw ║', ); console.error( '╚════════════════════════════════════════════════════════════════╝\n', ); throw new Error('Apple Container system is required but failed to start'); } } // Clean up stopped NanoClaw containers from previous runs try { const output = execSync('container ls -a --format {{.Names}}', { stdio: ['pipe', 'pipe', 'pipe'], encoding: 'utf-8', }); const stale = output .split('\n') .map((n) => n.trim()) .filter((n) => n.startsWith('nanoclaw-')); if (stale.length > 0) { execSync(`container rm ${stale.join(' ')}`, { stdio: 'pipe' }); logger.info({ count: stale.length }, 'Cleaned up stopped containers'); } } catch { // No stopped containers or ls/rm not supported } } async function main(): Promise { ensureContainerSystemRunning(); initDatabase(); logger.info('Database initialized'); loadState(); // Graceful shutdown handlers const shutdown = async (signal: string) => { logger.info({ signal }, 'Shutdown signal received'); await queue.shutdown(10000); process.exit(0); }; process.on('SIGTERM', () => shutdown('SIGTERM')); process.on('SIGINT', () => shutdown('SIGINT')); await connectWhatsApp(); } main().catch((err) => { logger.error({ err }, 'Failed to start NanoClaw'); process.exit(1); });