import fs from 'fs'; import path from 'path'; import { CronExpressionParser } from 'cron-parser'; import { DATA_DIR, IPC_POLL_INTERVAL, SERVICE_AGENT_TYPE, TIMEZONE, } from './config.js'; import { readJsonFile } from './utils.js'; import { AvailableGroup } from './agent-runner.js'; import { createTask, deleteTask, findDuplicateCiWatcher, getTaskById, updateTask, } from './db.js'; import { isValidGroupFolder } from './group-folder.js'; import { logger } from './logger.js'; import { DEFAULT_WATCH_CI_MAX_DURATION_MS, isWatchCiTask, } from './task-watch-status.js'; import { RegisteredGroup } from './types.js'; export interface IpcDeps { sendMessage: (jid: string, text: string) => Promise; nudgeScheduler?: () => void; registeredGroups: () => Record; registerGroup: (jid: string, group: RegisteredGroup) => void; syncGroups: (force: boolean) => Promise; getAvailableGroups: () => AvailableGroup[]; writeGroupsSnapshot: ( groupFolder: string, isMain: boolean, availableGroups: AvailableGroup[], ) => void; } let ipcWatcherRunning = false; const IPC_PROCESSING_DIRNAME = '.processing'; function buildIpcErrorPath( errorDir: string, prefix: string, fileName: string, ): string { return path.join(errorDir, `${prefix}-${Date.now()}-${fileName}`); } export function claimIpcFile(filePath: string): string | null { const processingDir = path.join( path.dirname(filePath), IPC_PROCESSING_DIRNAME, ); fs.mkdirSync(processingDir, { recursive: true }); const claimedPath = path.join(processingDir, path.basename(filePath)); try { fs.renameSync(filePath, claimedPath); return claimedPath; } catch (err) { if ((err as NodeJS.ErrnoException).code === 'ENOENT') { return null; } throw err; } } export function quarantineClaimedIpcFiles( ipcDir: string, errorDir: string, prefix: string, ): string[] { const processingDir = path.join(ipcDir, IPC_PROCESSING_DIRNAME); if (!fs.existsSync(processingDir)) { return []; } const movedPaths: string[] = []; for (const file of fs .readdirSync(processingDir) .filter((f) => f.endsWith('.json'))) { const claimedPath = path.join(processingDir, file); const errorPath = buildIpcErrorPath(errorDir, prefix, file); fs.renameSync(claimedPath, errorPath); movedPaths.push(errorPath); } return movedPaths; } function moveClaimedIpcFileToError( claimedPath: string, errorDir: string, prefix: string, ): void { fs.renameSync( claimedPath, buildIpcErrorPath(errorDir, prefix, path.basename(claimedPath)), ); } export function startIpcWatcher(deps: IpcDeps): 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; } const registeredGroups = deps.registeredGroups(); // Build folder→isMain lookup from registered groups const folderIsMain = new Map(); for (const group of Object.values(registeredGroups)) { if (group.isMain) folderIsMain.set(group.folder, true); } for (const sourceGroup of groupFolders) { const isMain = folderIsMain.get(sourceGroup) === true; const messagesDir = path.join(ipcBaseDir, sourceGroup, 'messages'); const tasksDir = path.join(ipcBaseDir, sourceGroup, 'tasks'); const errorDir = path.join(ipcBaseDir, 'errors'); fs.mkdirSync(errorDir, { recursive: true }); for (const quarantinedPath of quarantineClaimedIpcFiles( messagesDir, errorDir, `${sourceGroup}-message-stale`, )) { logger.warn( { sourceGroup, quarantinedPath }, 'Quarantined previously claimed IPC message after restart', ); } for (const quarantinedPath of quarantineClaimedIpcFiles( tasksDir, errorDir, `${sourceGroup}-task-stale`, )) { logger.warn( { sourceGroup, quarantinedPath }, 'Quarantined previously claimed IPC task after restart', ); } // 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); const claimedPath = claimIpcFile(filePath); if (!claimedPath) continue; try { const data = readJsonFile(claimedPath); if (!data || typeof data !== 'object') throw new Error('Invalid JSON'); const msg = data as { type?: string; chatJid?: string; text?: string; }; if (msg.type === 'message' && msg.chatJid && msg.text) { // Authorization: verify this group can send to this chatJid const targetGroup = registeredGroups[msg.chatJid]; const isMainOverride = isMain === true; if ( isMainOverride || (targetGroup && targetGroup.folder === sourceGroup) ) { await deps.sendMessage(msg.chatJid, msg.text); logger.info( { transition: 'ipc:auth:allow', chatJid: msg.chatJid, sourceGroup, targetGroup: targetGroup?.folder ?? null, isMainOverride, }, 'IPC message sent', ); } else { logger.warn( { transition: 'ipc:auth:deny', chatJid: msg.chatJid, sourceGroup, targetGroup: targetGroup?.folder ?? null, isMainOverride, }, 'Unauthorized IPC message attempt blocked', ); } } fs.unlinkSync(claimedPath); } catch (err) { logger.error( { file, sourceGroup, err }, 'Error processing IPC message', ); moveClaimedIpcFileToError( claimedPath, errorDir, `${sourceGroup}-message-error`, ); } } } } 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); const claimedPath = claimIpcFile(filePath); if (!claimedPath) continue; try { const data = readJsonFile(claimedPath); if (!data || typeof data !== 'object') throw new Error('Invalid JSON'); // Pass source group identity to processTaskIpc for authorization await processTaskIpc( data as Parameters[0], sourceGroup, isMain, deps, ); fs.unlinkSync(claimedPath); } catch (err) { logger.error( { file, sourceGroup, err }, 'Error processing IPC task', ); moveClaimedIpcFileToError( claimedPath, errorDir, `${sourceGroup}-task-error`, ); } } } } 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)'); } export async function processTaskIpc( data: { type: string; taskId?: string; prompt?: string; schedule_type?: string; schedule_value?: string; context_mode?: string; ci_provider?: 'github'; ci_metadata?: string; groupFolder?: string; chatJid?: string; targetJid?: string; // For register_group jid?: string; name?: string; folder?: string; trigger?: string; requiresTrigger?: boolean; agentConfig?: RegisteredGroup['agentConfig']; }, sourceGroup: string, // Verified identity from IPC directory isMain: boolean, // Verified from directory path deps: IpcDeps, ): Promise { const registeredGroups = deps.registeredGroups(); switch (data.type) { case 'schedule_task': if ( data.prompt && data.schedule_type && data.schedule_value && data.targetJid ) { // Resolve the target group from JID const targetJid = data.targetJid as string; const targetGroupEntry = registeredGroups[targetJid] || Object.values(registeredGroups).find( (group) => group.folder === targetJid, ); if (!targetGroupEntry) { logger.warn( { targetJid }, 'Cannot schedule task: target group not registered', ); break; } const targetFolder = targetGroupEntry.folder; const resolvedTargetJid = registeredGroups[targetJid] !== undefined ? targetJid : Object.entries(registeredGroups).find( ([, group]) => group.folder === targetFolder, )?.[0]; if (!resolvedTargetJid) { logger.warn( { targetJid, targetFolder }, 'Cannot resolve scheduled task target JID from folder', ); break; } // Authorization: non-main groups can only schedule for themselves if (!isMain && targetFolder !== sourceGroup) { logger.warn( { sourceGroup, targetFolder }, 'Unauthorized schedule_task attempt blocked', ); 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 = isWatchCiTask({ prompt: data.prompt }) ? new Date().toISOString() : new Date(Date.now() + ms).toISOString(); } else if (scheduleType === 'once') { const date = new Date(data.schedule_value); if (isNaN(date.getTime())) { logger.warn( { scheduleValue: data.schedule_value }, 'Invalid timestamp', ); break; } nextRun = date.toISOString(); } // Deduplicate CI watchers: if another agent already watches the same // channel + provider + run, skip creation to avoid duplicate notifications. if (data.ci_provider && data.ci_metadata) { const existing = findDuplicateCiWatcher( resolvedTargetJid, data.ci_provider, data.ci_metadata as string, ); if (existing) { logger.info( { existingTaskId: existing.id, existingAgentType: existing.agent_type, ciProvider: data.ci_provider, sourceGroup, }, 'Duplicate CI watcher skipped — another agent already watches this run', ); break; } } const taskId = data.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: targetFolder, chat_jid: resolvedTargetJid, agent_type: targetGroupEntry.agentType || SERVICE_AGENT_TYPE, ci_provider: data.ci_provider ?? null, ci_metadata: data.ci_metadata ?? null, max_duration_ms: isWatchCiTask({ prompt: data.prompt }) ? DEFAULT_WATCH_CI_MAX_DURATION_MS : null, 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, targetFolder, contextMode, agentType: targetGroupEntry.agentType || SERVICE_AGENT_TYPE, }, 'Task created via IPC', ); if (nextRun && new Date(nextRun).getTime() <= Date.now()) { deps.nudgeScheduler?.(); } } 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 'update_task': if (data.taskId) { const task = getTaskById(data.taskId); if (!task) { logger.warn( { taskId: data.taskId, sourceGroup }, 'Task not found for update', ); break; } if (!isMain && task.group_folder !== sourceGroup) { logger.warn( { taskId: data.taskId, sourceGroup }, 'Unauthorized task update attempt', ); break; } const updates: Parameters[1] = {}; if (data.prompt !== undefined) updates.prompt = data.prompt; if (data.schedule_type !== undefined) updates.schedule_type = data.schedule_type as | 'cron' | 'interval' | 'once'; if (data.schedule_value !== undefined) updates.schedule_value = data.schedule_value; // Recompute next_run if schedule changed if (data.schedule_type || data.schedule_value) { const updatedTask = { ...task, ...updates, }; if (updatedTask.schedule_type === 'cron') { try { const interval = CronExpressionParser.parse( updatedTask.schedule_value, { tz: TIMEZONE }, ); updates.next_run = interval.next().toISOString(); } catch { logger.warn( { taskId: data.taskId, value: updatedTask.schedule_value }, 'Invalid cron in task update', ); break; } } else if (updatedTask.schedule_type === 'interval') { const ms = parseInt(updatedTask.schedule_value, 10); if (!isNaN(ms) && ms > 0) { updates.next_run = new Date(Date.now() + ms).toISOString(); } } } updateTask(data.taskId, updates); logger.info( { taskId: data.taskId, sourceGroup, updates }, 'Task updated via IPC', ); } break; case 'refresh_groups': // Only main group can request a refresh if (isMain) { logger.info( { sourceGroup }, 'Group metadata refresh requested via IPC', ); await deps.syncGroups(true); // Write updated snapshot immediately const availableGroups = deps.getAvailableGroups(); deps.writeGroupsSnapshot(sourceGroup, true, availableGroups); } 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) { if (!isValidGroupFolder(data.folder)) { logger.warn( { sourceGroup, folder: data.folder }, 'Invalid register_group request - unsafe folder name', ); break; } // Defense in depth: agent cannot set isMain via IPC deps.registerGroup(data.jid, { name: data.name, folder: data.folder, trigger: data.trigger, added_at: new Date().toISOString(), agentConfig: data.agentConfig, requiresTrigger: data.requiresTrigger, }); } else { logger.warn( { data }, 'Invalid register_group request - missing required fields', ); } break; default: logger.warn({ type: data.type }, 'Unknown IPC task type'); } }