diff --git a/quality/code-quality-budgets.json b/quality/code-quality-budgets.json index b40aeaf..7f2fa21 100644 --- a/quality/code-quality-budgets.json +++ b/quality/code-quality-budgets.json @@ -109,11 +109,6 @@ "maxFunctionLines": 314, "owner": "main service entrypoint hotspot" }, - "src/ipc-task-processor.ts": { - "maxFunctionLines": 566, - "maxComplexity": 107, - "owner": "ipc task processor hotspot" - }, "src/message-agent-executor-attempt-runner.ts": { "maxFunctionLines": 275, "maxComplexity": 47, diff --git a/src/ipc-auth.test.ts b/src/ipc-auth.test.ts index f262f91..cc8b019 100644 --- a/src/ipc-auth.test.ts +++ b/src/ipc-auth.test.ts @@ -384,6 +384,42 @@ describe('resume_task authorization', () => { }); }); +// --- update_task behavior --- + +describe('update_task behavior', () => { + it('updates once tasks even when next_run is not recomputed', async () => { + createTask({ + id: 'task-once-update', + group_folder: 'other-group', + chat_jid: 'other@g.us', + prompt: 'original once task', + schedule_type: 'once', + schedule_value: '2025-06-01T00:00:00', + context_mode: 'isolated', + next_run: '2025-06-01T00:00:00.000Z', + status: 'active', + created_at: '2024-01-01T00:00:00.000Z', + }); + + await processTaskIpc( + { + type: 'update_task', + taskId: 'task-once-update', + prompt: 'updated once task', + schedule_value: '2025-07-01T00:00:00', + }, + 'other-group', + false, + deps, + ); + + const task = getTaskById('task-once-update')!; + expect(task.prompt).toBe('updated once task'); + expect(task.schedule_value).toBe('2025-07-01T00:00:00'); + expect(task.next_run).toBe('2025-06-01T00:00:00.000Z'); + }); +}); + // --- cancel_task authorization --- describe('cancel_task authorization', () => { diff --git a/src/ipc-task-processor.ts b/src/ipc-task-processor.ts index 7ec281d..b1f56db 100644 --- a/src/ipc-task-processor.ts +++ b/src/ipc-task-processor.ts @@ -1,41 +1,17 @@ -import { CronExpressionParser } from 'cron-parser'; - -import { TIMEZONE } from './config.js'; -import { - isHostEvidenceAction, - runHostEvidenceRequest, - writeHostEvidenceResponse, -} from './host-evidence.js'; -import { - createTask, - deleteTask, - findDuplicateCiWatcher, - getTaskById, - rememberMemory, - updateTask, -} from './db.js'; -import { isValidGroupFolder } from './group-folder.js'; +import { deleteTask, updateTask } from './db.js'; import { logger } from './logger.js'; -import { normalizeStoredAgentType } from './db/room-registration.js'; +import { handleHostEvidenceRequest } from './ipc-task-processor/host-evidence.js'; +import { handlePersistMemory } from './ipc-task-processor/memory.js'; import { - DEFAULT_WATCH_CI_MAX_DURATION_MS, - isWatchCiTask, -} from './task-watch-status.js'; + handleAssignRoom, + handleRefreshGroups, +} from './ipc-task-processor/rooms.js'; +import { handleScheduleTask } from './ipc-task-processor/schedule.js'; +import { + handleTaskStateMutation, + handleUpdateTask, +} from './ipc-task-processor/tasks.js'; import type { IpcDeps, TaskIpcPayload } from './ipc-types.js'; -import type { PairedRoomRole } from './types.js'; - -type RoomBindings = ReturnType; -type ScheduleType = 'cron' | 'interval' | 'once'; -type TaskUpdates = Parameters[1]; -type MemorySourceKind = Parameters[0]['sourceKind']; - -function normalizePairedRoomRole( - role: string | null | undefined, -): PairedRoomRole | undefined { - return role === 'owner' || role === 'reviewer' || role === 'arbiter' - ? role - : undefined; -} export async function processTaskIpc( data: TaskIpcPayload, @@ -43,11 +19,9 @@ export async function processTaskIpc( isMain: boolean, deps: IpcDeps, ): Promise { - const roomBindings = deps.roomBindings(); - switch (data.type) { case 'schedule_task': - handleScheduleTask(data, sourceGroup, isMain, deps, roomBindings); + handleScheduleTask(data, sourceGroup, isMain, deps); break; case 'pause_task': @@ -98,569 +72,3 @@ export async function processTaskIpc( logger.warn({ type: data.type }, 'Unknown IPC task type'); } } - -function handleScheduleTask( - data: TaskIpcPayload, - sourceGroup: string, - isMain: boolean, - deps: IpcDeps, - roomBindings: RoomBindings, -): void { - if ( - !data.prompt || - !data.schedule_type || - !data.schedule_value || - !data.targetJid - ) { - return; - } - - const targetJid = data.targetJid; - const targetGroupEntry = - roomBindings[targetJid] || - Object.values(roomBindings).find((group) => group.folder === targetJid); - - if (!targetGroupEntry) { - logger.warn( - { targetJid }, - 'Cannot schedule task: target group not registered', - ); - return; - } - - const targetFolder = targetGroupEntry.folder; - const resolvedTargetJid = - roomBindings[targetJid] !== undefined - ? targetJid - : Object.entries(roomBindings).find( - ([, group]) => group.folder === targetFolder, - )?.[0]; - - if (!resolvedTargetJid) { - logger.warn( - { targetJid, targetFolder }, - 'Cannot resolve scheduled task target JID from folder', - ); - return; - } - - if (!isMain && targetFolder !== sourceGroup) { - logger.warn( - { sourceGroup, targetFolder }, - 'Unauthorized schedule_task attempt blocked', - ); - return; - } - - const scheduleType = data.schedule_type as ScheduleType; - const nextRunResult = resolveNextRun(data, scheduleType); - if (!nextRunResult.ok) return; - - if (data.ci_provider && data.ci_metadata) { - const existing = findDuplicateCiWatcher( - resolvedTargetJid, - data.ci_provider, - data.ci_metadata, - ); - 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', - ); - return; - } - } - - 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'; - const scheduledAgentType = - normalizeStoredAgentType(data.agent_type) ?? - targetGroupEntry.agentType ?? - 'claude-code'; - const scheduledRoomRole = normalizePairedRoomRole(data.room_role); - createTask({ - id: taskId, - group_folder: targetFolder, - chat_jid: resolvedTargetJid, - agent_type: scheduledAgentType, - room_role: scheduledRoomRole ?? null, - 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: nextRunResult.nextRun, - status: 'active', - created_at: new Date().toISOString(), - }); - logger.info( - { - taskId, - sourceGroup, - targetFolder, - contextMode, - agentType: scheduledAgentType, - roomRole: scheduledRoomRole ?? null, - }, - 'Task created via IPC', - ); - if ( - nextRunResult.nextRun && - new Date(nextRunResult.nextRun).getTime() <= Date.now() - ) { - deps.nudgeScheduler?.(); - } -} - -function resolveNextRun( - data: TaskIpcPayload, - scheduleType: ScheduleType, -): { ok: true; nextRun: string | null } | { ok: false } { - if (scheduleType === 'cron') { - try { - const interval = CronExpressionParser.parse(data.schedule_value!, { - tz: TIMEZONE, - }); - return { ok: true, nextRun: interval.next().toISOString() }; - } catch { - logger.warn( - { scheduleValue: data.schedule_value }, - 'Invalid cron expression', - ); - return { ok: false }; - } - } - - if (scheduleType === 'interval') { - const ms = parseInt(data.schedule_value!, 10); - if (isNaN(ms) || ms <= 0) { - logger.warn({ scheduleValue: data.schedule_value }, 'Invalid interval'); - return { ok: false }; - } - return { - ok: true, - nextRun: isWatchCiTask({ prompt: data.prompt! }) - ? new Date().toISOString() - : new Date(Date.now() + ms).toISOString(), - }; - } - - if (scheduleType === 'once') { - const date = new Date(data.schedule_value!); - if (isNaN(date.getTime())) { - logger.warn({ scheduleValue: data.schedule_value }, 'Invalid timestamp'); - return { ok: false }; - } - return { ok: true, nextRun: date.toISOString() }; - } - - return { ok: true, nextRun: null }; -} - -function handleTaskStateMutation( - data: TaskIpcPayload, - sourceGroup: string, - isMain: boolean, - config: { - action: () => void; - successMessage: string; - unauthorizedMessage: string; - }, -): void { - if (!data.taskId) return; - - const task = getTaskById(data.taskId); - if (task && (isMain || task.group_folder === sourceGroup)) { - config.action(); - logger.info({ taskId: data.taskId, sourceGroup }, config.successMessage); - } else { - logger.warn( - { taskId: data.taskId, sourceGroup }, - config.unauthorizedMessage, - ); - } -} - -async function handleHostEvidenceRequest( - data: TaskIpcPayload, - sourceGroup: string, - isMain: boolean, - deps: IpcDeps, -): Promise { - if (!data.requestId) { - logger.warn( - { sourceGroup }, - 'Ignoring host_evidence_request without requestId', - ); - return; - } - - if (data.action === 'ejclaw_room_runtime') { - handleRoomRuntimeReportRequest(data, sourceGroup, isMain, deps); - return; - } - - if (!isHostEvidenceAction(data.action)) { - writeHostEvidenceResponse(sourceGroup, { - requestId: data.requestId, - ok: false, - action: 'ejclaw_service_status', - command: '', - stdout: '', - stderr: '', - exitCode: 1, - error: `Unsupported host evidence action: ${String(data.action)}`, - }); - logger.warn( - { sourceGroup, requestId: data.requestId, action: data.action }, - 'Rejected unsupported host evidence action', - ); - return; - } - - const result = await runHostEvidenceRequest({ - requestId: data.requestId, - action: data.action, - tailLines: - typeof data.tail_lines === 'number' ? data.tail_lines : undefined, - taskId: typeof data.task_id === 'string' ? data.task_id : undefined, - minutes: typeof data.minutes === 'number' ? data.minutes : undefined, - limit: typeof data.limit === 'number' ? data.limit : undefined, - repo: typeof data.repo === 'string' ? data.repo : undefined, - prNumber: typeof data.pr_number === 'number' ? data.pr_number : undefined, - runId: typeof data.run_id === 'number' ? data.run_id : undefined, - artifactKind: - typeof data.artifact_kind === 'string' ? data.artifact_kind : undefined, - sourceGroup, - isMain, - }); - - writeHostEvidenceResponse(sourceGroup, { - requestId: data.requestId, - ...result, - }); - - logger.info( - { - sourceGroup, - requestId: data.requestId, - action: data.action, - ok: result.ok, - exitCode: result.exitCode, - }, - 'Processed host evidence request via IPC', - ); -} - -function handleRoomRuntimeReportRequest( - data: TaskIpcPayload, - sourceGroup: string, - isMain: boolean, - deps: IpcDeps, -): void { - if (!data.chatJid) { - writeHostEvidenceResponse(sourceGroup, { - requestId: data.requestId!, - ok: false, - action: 'ejclaw_room_runtime', - command: 'internal:ejclaw_room_runtime', - stdout: '', - stderr: '', - exitCode: 1, - error: 'Missing chatJid for ejclaw_room_runtime request', - }); - logger.warn( - { sourceGroup, requestId: data.requestId }, - 'Rejected ejclaw_room_runtime request without chatJid', - ); - return; - } - - if (!deps.getRoomRuntimeReport) { - writeHostEvidenceResponse(sourceGroup, { - requestId: data.requestId!, - ok: false, - action: 'ejclaw_room_runtime', - command: 'internal:ejclaw_room_runtime', - stdout: '', - stderr: '', - exitCode: 1, - error: 'Room runtime reporting is not configured', - }); - logger.warn( - { sourceGroup, requestId: data.requestId, chatJid: data.chatJid }, - 'Rejected ejclaw_room_runtime request because runtime reporter is unavailable', - ); - return; - } - - const report = deps.getRoomRuntimeReport({ - chatJid: data.chatJid, - sourceGroup, - isMain, - }); - writeHostEvidenceResponse(sourceGroup, { - requestId: data.requestId!, - ok: true, - action: 'ejclaw_room_runtime', - command: 'internal:ejclaw_room_runtime', - stdout: JSON.stringify(report, null, 2), - stderr: '', - exitCode: 0, - }); - logger.info( - { - sourceGroup, - requestId: data.requestId, - action: data.action, - chatJid: data.chatJid, - }, - 'Processed room runtime report request via IPC', - ); -} - -function handleUpdateTask( - data: TaskIpcPayload, - sourceGroup: string, - isMain: boolean, -): void { - if (!data.taskId) return; - - const task = getTaskById(data.taskId); - if (!task) { - logger.warn( - { taskId: data.taskId, sourceGroup }, - 'Task not found for update', - ); - return; - } - if (!isMain && task.group_folder !== sourceGroup) { - logger.warn( - { taskId: data.taskId, sourceGroup }, - 'Unauthorized task update attempt', - ); - return; - } - - const updates: TaskUpdates = {}; - if (data.prompt !== undefined) updates.prompt = data.prompt; - if (data.schedule_type !== undefined) { - updates.schedule_type = data.schedule_type as ScheduleType; - } - if (data.schedule_value !== undefined) { - updates.schedule_value = data.schedule_value; - } - - 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', - ); - return; - } - } 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', - ); -} - -async function handleRefreshGroups( - sourceGroup: string, - isMain: boolean, - deps: IpcDeps, -): Promise { - if (!isMain) { - logger.warn({ sourceGroup }, 'Unauthorized refresh_groups attempt blocked'); - return; - } - - logger.info({ sourceGroup }, 'Group metadata refresh requested via IPC'); - await deps.syncGroups(true); - const availableGroups = deps.getAvailableGroups(); - deps.writeGroupsSnapshot(sourceGroup, true, availableGroups); -} - -function handleAssignRoom( - data: TaskIpcPayload, - sourceGroup: string, - isMain: boolean, - deps: IpcDeps, -): void { - if (!isMain) { - logger.warn( - { sourceGroup, type: data.type }, - `Unauthorized ${data.type} attempt blocked`, - ); - return; - } - if (!data.jid || !data.name) { - logger.warn( - { data }, - `Invalid ${data.type} request - missing required fields`, - ); - return; - } - if (data.folder && !isValidGroupFolder(data.folder)) { - logger.warn( - { sourceGroup, folder: data.folder }, - `Invalid ${data.type} request - unsafe folder name`, - ); - return; - } - if ( - data.room_mode !== undefined && - data.room_mode !== 'single' && - data.room_mode !== 'tribunal' - ) { - logger.warn( - { sourceGroup, roomMode: data.room_mode }, - 'Invalid assign_room request - unknown room_mode', - ); - return; - } - if ( - data.owner_agent_type !== undefined && - data.owner_agent_type !== 'claude-code' && - data.owner_agent_type !== 'codex' - ) { - logger.warn( - { sourceGroup, ownerAgentType: data.owner_agent_type }, - 'Invalid assign_room request - unknown owner_agent_type', - ); - return; - } - if ( - data.reviewer_agent_type !== undefined && - data.reviewer_agent_type !== 'claude-code' && - data.reviewer_agent_type !== 'codex' - ) { - logger.warn( - { sourceGroup, reviewerAgentType: data.reviewer_agent_type }, - 'Invalid assign_room request - unknown reviewer_agent_type', - ); - return; - } - if ( - data.arbiter_agent_type !== undefined && - data.arbiter_agent_type !== null && - data.arbiter_agent_type !== 'claude-code' && - data.arbiter_agent_type !== 'codex' - ) { - logger.warn( - { sourceGroup, arbiterAgentType: data.arbiter_agent_type }, - 'Invalid assign_room request - unknown arbiter_agent_type', - ); - return; - } - - deps.assignRoom(data.jid, { - name: data.name, - roomMode: data.room_mode, - ownerAgentType: data.owner_agent_type, - reviewerAgentType: data.reviewer_agent_type, - arbiterAgentType: data.arbiter_agent_type, - folder: data.folder, - isMain: data.isMain, - workDir: data.workDir, - }); -} - -function handlePersistMemory(data: TaskIpcPayload, sourceGroup: string): void { - if ( - data.scopeKind !== 'room' || - typeof data.scopeKey !== 'string' || - typeof data.content !== 'string' - ) { - logger.warn( - { sourceGroup, data }, - 'Invalid persist_memory request - missing required fields', - ); - return; - } - - const expectedScopeKey = `room:${sourceGroup}`; - if (data.scopeKey !== expectedScopeKey) { - logger.warn( - { sourceGroup, scopeKey: data.scopeKey, expectedScopeKey }, - 'Unauthorized persist_memory attempt blocked', - ); - return; - } - - if ( - data.source_kind !== undefined && - data.source_kind !== 'compact' && - data.source_kind !== 'explicit' && - data.source_kind !== 'import' && - data.source_kind !== 'system' - ) { - logger.warn( - { sourceGroup, sourceKind: data.source_kind }, - 'Invalid persist_memory request - unknown source_kind', - ); - return; - } - - if ( - Array.isArray(data.keywords) && - !data.keywords.every((value) => typeof value === 'string') - ) { - logger.warn( - { sourceGroup, keywords: data.keywords }, - 'Invalid persist_memory request - keywords must be strings', - ); - return; - } - - rememberMemory({ - scopeKind: 'room', - scopeKey: data.scopeKey, - content: data.content, - keywords: data.keywords, - memoryKind: typeof data.memory_kind === 'string' ? data.memory_kind : null, - sourceKind: (data.source_kind as MemorySourceKind | undefined) ?? 'compact', - sourceRef: typeof data.source_ref === 'string' ? data.source_ref : null, - }); - logger.info( - { - sourceGroup, - scopeKey: data.scopeKey, - sourceKind: data.source_kind ?? 'compact', - }, - 'Memory persisted via IPC', - ); -} diff --git a/src/ipc-task-processor/host-evidence.ts b/src/ipc-task-processor/host-evidence.ts new file mode 100644 index 0000000..7e96281 --- /dev/null +++ b/src/ipc-task-processor/host-evidence.ts @@ -0,0 +1,145 @@ +import { + isHostEvidenceAction, + runHostEvidenceRequest, + writeHostEvidenceResponse, +} from '../host-evidence.js'; +import { logger } from '../logger.js'; +import type { IpcDeps, TaskIpcPayload } from '../ipc-types.js'; + +export async function handleHostEvidenceRequest( + data: TaskIpcPayload, + sourceGroup: string, + isMain: boolean, + deps: IpcDeps, +): Promise { + if (!data.requestId) { + logger.warn( + { sourceGroup }, + 'Ignoring host_evidence_request without requestId', + ); + return; + } + + if (data.action === 'ejclaw_room_runtime') { + handleRoomRuntimeReportRequest(data, sourceGroup, isMain, deps); + return; + } + + if (!isHostEvidenceAction(data.action)) { + writeHostEvidenceResponse(sourceGroup, { + requestId: data.requestId, + ok: false, + action: 'ejclaw_service_status', + command: '', + stdout: '', + stderr: '', + exitCode: 1, + error: `Unsupported host evidence action: ${String(data.action)}`, + }); + logger.warn( + { sourceGroup, requestId: data.requestId, action: data.action }, + 'Rejected unsupported host evidence action', + ); + return; + } + + const result = await runHostEvidenceRequest({ + requestId: data.requestId, + action: data.action, + tailLines: + typeof data.tail_lines === 'number' ? data.tail_lines : undefined, + taskId: typeof data.task_id === 'string' ? data.task_id : undefined, + minutes: typeof data.minutes === 'number' ? data.minutes : undefined, + limit: typeof data.limit === 'number' ? data.limit : undefined, + repo: typeof data.repo === 'string' ? data.repo : undefined, + prNumber: typeof data.pr_number === 'number' ? data.pr_number : undefined, + runId: typeof data.run_id === 'number' ? data.run_id : undefined, + artifactKind: + typeof data.artifact_kind === 'string' ? data.artifact_kind : undefined, + sourceGroup, + isMain, + }); + + writeHostEvidenceResponse(sourceGroup, { + requestId: data.requestId, + ...result, + }); + + logger.info( + { + sourceGroup, + requestId: data.requestId, + action: data.action, + ok: result.ok, + exitCode: result.exitCode, + }, + 'Processed host evidence request via IPC', + ); +} + +function handleRoomRuntimeReportRequest( + data: TaskIpcPayload, + sourceGroup: string, + isMain: boolean, + deps: IpcDeps, +): void { + if (!data.chatJid) { + writeHostEvidenceResponse(sourceGroup, { + requestId: data.requestId!, + ok: false, + action: 'ejclaw_room_runtime', + command: 'internal:ejclaw_room_runtime', + stdout: '', + stderr: '', + exitCode: 1, + error: 'Missing chatJid for ejclaw_room_runtime request', + }); + logger.warn( + { sourceGroup, requestId: data.requestId }, + 'Rejected ejclaw_room_runtime request without chatJid', + ); + return; + } + + if (!deps.getRoomRuntimeReport) { + writeHostEvidenceResponse(sourceGroup, { + requestId: data.requestId!, + ok: false, + action: 'ejclaw_room_runtime', + command: 'internal:ejclaw_room_runtime', + stdout: '', + stderr: '', + exitCode: 1, + error: 'Room runtime reporting is not configured', + }); + logger.warn( + { sourceGroup, requestId: data.requestId, chatJid: data.chatJid }, + 'Rejected ejclaw_room_runtime request because runtime reporter is unavailable', + ); + return; + } + + const report = deps.getRoomRuntimeReport({ + chatJid: data.chatJid, + sourceGroup, + isMain, + }); + writeHostEvidenceResponse(sourceGroup, { + requestId: data.requestId!, + ok: true, + action: 'ejclaw_room_runtime', + command: 'internal:ejclaw_room_runtime', + stdout: JSON.stringify(report, null, 2), + stderr: '', + exitCode: 0, + }); + logger.info( + { + sourceGroup, + requestId: data.requestId, + action: data.action, + chatJid: data.chatJid, + }, + 'Processed room runtime report request via IPC', + ); +} diff --git a/src/ipc-task-processor/memory.ts b/src/ipc-task-processor/memory.ts new file mode 100644 index 0000000..375e256 --- /dev/null +++ b/src/ipc-task-processor/memory.ts @@ -0,0 +1,96 @@ +import { rememberMemory } from '../db.js'; +import { logger } from '../logger.js'; +import type { TaskIpcPayload } from '../ipc-types.js'; + +type MemorySourceKind = Parameters[0]['sourceKind']; + +export function handlePersistMemory( + data: TaskIpcPayload, + sourceGroup: string, +): void { + if (!hasPersistMemoryFields(data)) { + logger.warn( + { sourceGroup, data }, + 'Invalid persist_memory request - missing required fields', + ); + return; + } + + const expectedScopeKey = `room:${sourceGroup}`; + if (data.scopeKey !== expectedScopeKey) { + logger.warn( + { sourceGroup, scopeKey: data.scopeKey, expectedScopeKey }, + 'Unauthorized persist_memory attempt blocked', + ); + return; + } + + if (!isMemorySourceKind(data.source_kind)) { + logger.warn( + { sourceGroup, sourceKind: data.source_kind }, + 'Invalid persist_memory request - unknown source_kind', + ); + return; + } + + if (!hasValidKeywords(data.keywords)) { + logger.warn( + { sourceGroup, keywords: data.keywords }, + 'Invalid persist_memory request - keywords must be strings', + ); + return; + } + + rememberMemory({ + scopeKind: 'room', + scopeKey: data.scopeKey, + content: data.content, + keywords: data.keywords, + memoryKind: typeof data.memory_kind === 'string' ? data.memory_kind : null, + sourceKind: data.source_kind ?? 'compact', + sourceRef: typeof data.source_ref === 'string' ? data.source_ref : null, + }); + logger.info( + { + sourceGroup, + scopeKey: data.scopeKey, + sourceKind: data.source_kind ?? 'compact', + }, + 'Memory persisted via IPC', + ); +} + +function hasPersistMemoryFields( + data: TaskIpcPayload, +): data is TaskIpcPayload & { + scopeKind: 'room'; + scopeKey: string; + content: string; +} { + return ( + data.scopeKind === 'room' && + typeof data.scopeKey === 'string' && + typeof data.content === 'string' + ); +} + +function isMemorySourceKind( + value: string | undefined, +): value is MemorySourceKind | undefined { + return ( + value === undefined || + value === 'compact' || + value === 'explicit' || + value === 'import' || + value === 'system' + ); +} + +function hasValidKeywords( + value: TaskIpcPayload['keywords'], +): value is string[] | undefined { + return ( + value === undefined || + (Array.isArray(value) && value.every((item) => typeof item === 'string')) + ); +} diff --git a/src/ipc-task-processor/rooms.ts b/src/ipc-task-processor/rooms.ts new file mode 100644 index 0000000..b182ec5 --- /dev/null +++ b/src/ipc-task-processor/rooms.ts @@ -0,0 +1,116 @@ +import { isValidGroupFolder } from '../group-folder.js'; +import { logger } from '../logger.js'; +import type { IpcDeps, TaskIpcPayload } from '../ipc-types.js'; +import type { AgentType, RoomMode } from '../types.js'; + +export async function handleRefreshGroups( + sourceGroup: string, + isMain: boolean, + deps: IpcDeps, +): Promise { + if (!isMain) { + logger.warn({ sourceGroup }, 'Unauthorized refresh_groups attempt blocked'); + return; + } + + logger.info({ sourceGroup }, 'Group metadata refresh requested via IPC'); + await deps.syncGroups(true); + const availableGroups = deps.getAvailableGroups(); + deps.writeGroupsSnapshot(sourceGroup, true, availableGroups); +} + +export function handleAssignRoom( + data: TaskIpcPayload, + sourceGroup: string, + isMain: boolean, + deps: IpcDeps, +): void { + const validation = validateAssignRoomRequest(data, sourceGroup, isMain); + if (!validation.ok) return; + + deps.assignRoom(data.jid!, { + name: data.name!, + roomMode: data.room_mode, + ownerAgentType: data.owner_agent_type, + reviewerAgentType: data.reviewer_agent_type, + arbiterAgentType: data.arbiter_agent_type, + folder: data.folder, + isMain: data.isMain, + workDir: data.workDir, + }); +} + +function validateAssignRoomRequest( + data: TaskIpcPayload, + sourceGroup: string, + isMain: boolean, +): { ok: true } | { ok: false } { + if (!isMain) { + logger.warn( + { sourceGroup, type: data.type }, + `Unauthorized ${data.type} attempt blocked`, + ); + return { ok: false }; + } + if (!data.jid || !data.name) { + logger.warn( + { data }, + `Invalid ${data.type} request - missing required fields`, + ); + return { ok: false }; + } + if (data.folder && !isValidGroupFolder(data.folder)) { + logger.warn( + { sourceGroup, folder: data.folder }, + `Invalid ${data.type} request - unsafe folder name`, + ); + return { ok: false }; + } + if (data.room_mode !== undefined && !isRoomMode(data.room_mode)) { + logger.warn( + { sourceGroup, roomMode: data.room_mode }, + 'Invalid assign_room request - unknown room_mode', + ); + return { ok: false }; + } + if ( + data.owner_agent_type !== undefined && + !isAssignableAgentType(data.owner_agent_type) + ) { + logger.warn( + { sourceGroup, ownerAgentType: data.owner_agent_type }, + 'Invalid assign_room request - unknown owner_agent_type', + ); + return { ok: false }; + } + if ( + data.reviewer_agent_type !== undefined && + !isAssignableAgentType(data.reviewer_agent_type) + ) { + logger.warn( + { sourceGroup, reviewerAgentType: data.reviewer_agent_type }, + 'Invalid assign_room request - unknown reviewer_agent_type', + ); + return { ok: false }; + } + if ( + data.arbiter_agent_type !== undefined && + data.arbiter_agent_type !== null && + !isAssignableAgentType(data.arbiter_agent_type) + ) { + logger.warn( + { sourceGroup, arbiterAgentType: data.arbiter_agent_type }, + 'Invalid assign_room request - unknown arbiter_agent_type', + ); + return { ok: false }; + } + return { ok: true }; +} + +function isRoomMode(value: string): value is RoomMode { + return value === 'single' || value === 'tribunal'; +} + +function isAssignableAgentType(value: string): value is AgentType { + return value === 'claude-code' || value === 'codex'; +} diff --git a/src/ipc-task-processor/schedule.ts b/src/ipc-task-processor/schedule.ts new file mode 100644 index 0000000..12ec9ec --- /dev/null +++ b/src/ipc-task-processor/schedule.ts @@ -0,0 +1,228 @@ +import { CronExpressionParser } from 'cron-parser'; + +import { createTask, findDuplicateCiWatcher } from '../db.js'; +import { normalizeStoredAgentType } from '../db/room-registration.js'; +import { TIMEZONE } from '../config.js'; +import { logger } from '../logger.js'; +import { + DEFAULT_WATCH_CI_MAX_DURATION_MS, + isWatchCiTask, +} from '../task-watch-status.js'; +import type { IpcDeps, TaskIpcPayload } from '../ipc-types.js'; +import type { PairedRoomRole, RegisteredGroup } from '../types.js'; + +type RoomBindings = ReturnType; +type ScheduleType = 'cron' | 'interval' | 'once'; + +interface ResolvedScheduleTarget { + folder: string; + jid: string; + room: RegisteredGroup; +} + +function normalizePairedRoomRole( + role: string | null | undefined, +): PairedRoomRole | undefined { + return role === 'owner' || role === 'reviewer' || role === 'arbiter' + ? role + : undefined; +} + +export function handleScheduleTask( + data: TaskIpcPayload, + sourceGroup: string, + isMain: boolean, + deps: IpcDeps, +): void { + if (!hasRequiredScheduleFields(data)) return; + + const roomBindings = deps.roomBindings(); + const target = resolveScheduleTarget(data.targetJid, roomBindings); + if (!target) return; + + if (!isMain && target.folder !== sourceGroup) { + logger.warn( + { sourceGroup, targetFolder: target.folder }, + 'Unauthorized schedule_task attempt blocked', + ); + return; + } + + const scheduleType = data.schedule_type as ScheduleType; + const nextRunResult = resolveNextRun(data, scheduleType); + if (!nextRunResult.ok) return; + + if (isDuplicateCiWatcher(data, target.jid, sourceGroup)) return; + + 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'; + const scheduledAgentType = + normalizeStoredAgentType(data.agent_type) ?? + target.room.agentType ?? + 'claude-code'; + const scheduledRoomRole = normalizePairedRoomRole(data.room_role); + + createTask({ + id: taskId, + group_folder: target.folder, + chat_jid: target.jid, + agent_type: scheduledAgentType, + room_role: scheduledRoomRole ?? null, + 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: nextRunResult.nextRun, + status: 'active', + created_at: new Date().toISOString(), + }); + logger.info( + { + taskId, + sourceGroup, + targetFolder: target.folder, + contextMode, + agentType: scheduledAgentType, + roomRole: scheduledRoomRole ?? null, + }, + 'Task created via IPC', + ); + if ( + nextRunResult.nextRun && + new Date(nextRunResult.nextRun).getTime() <= Date.now() + ) { + deps.nudgeScheduler?.(); + } +} + +function hasRequiredScheduleFields( + data: TaskIpcPayload, +): data is TaskIpcPayload & { + prompt: string; + schedule_type: string; + schedule_value: string; + targetJid: string; +} { + return Boolean( + data.prompt && data.schedule_type && data.schedule_value && data.targetJid, + ); +} + +function resolveScheduleTarget( + targetJid: string, + roomBindings: RoomBindings, +): ResolvedScheduleTarget | null { + const targetRoom = + roomBindings[targetJid] || + Object.values(roomBindings).find((group) => group.folder === targetJid); + + if (!targetRoom) { + logger.warn( + { targetJid }, + 'Cannot schedule task: target group not registered', + ); + return null; + } + + const resolvedTargetJid = + roomBindings[targetJid] !== undefined + ? targetJid + : Object.entries(roomBindings).find( + ([, group]) => group.folder === targetRoom.folder, + )?.[0]; + + if (!resolvedTargetJid) { + logger.warn( + { targetJid, targetFolder: targetRoom.folder }, + 'Cannot resolve scheduled task target JID from folder', + ); + return null; + } + + return { + folder: targetRoom.folder, + jid: resolvedTargetJid, + room: targetRoom, + }; +} + +function isDuplicateCiWatcher( + data: TaskIpcPayload, + targetJid: string, + sourceGroup: string, +): boolean { + if (!data.ci_provider || !data.ci_metadata) return false; + + const existing = findDuplicateCiWatcher( + targetJid, + data.ci_provider, + data.ci_metadata, + ); + if (!existing) return false; + + logger.info( + { + existingTaskId: existing.id, + existingAgentType: existing.agent_type, + ciProvider: data.ci_provider, + sourceGroup, + }, + 'Duplicate CI watcher skipped — another agent already watches this run', + ); + return true; +} + +function resolveNextRun( + data: TaskIpcPayload & { prompt?: string; schedule_value?: string }, + scheduleType: ScheduleType, +): { ok: true; nextRun: string | null } | { ok: false } { + if (scheduleType === 'cron') { + try { + const interval = CronExpressionParser.parse(data.schedule_value!, { + tz: TIMEZONE, + }); + return { ok: true, nextRun: interval.next().toISOString() }; + } catch { + logger.warn( + { scheduleValue: data.schedule_value }, + 'Invalid cron expression', + ); + return { ok: false }; + } + } + + if (scheduleType === 'interval') { + const ms = parseInt(data.schedule_value!, 10); + if (isNaN(ms) || ms <= 0) { + logger.warn({ scheduleValue: data.schedule_value }, 'Invalid interval'); + return { ok: false }; + } + return { + ok: true, + nextRun: isWatchCiTask({ prompt: data.prompt! }) + ? new Date().toISOString() + : new Date(Date.now() + ms).toISOString(), + }; + } + + if (scheduleType === 'once') { + const date = new Date(data.schedule_value!); + if (isNaN(date.getTime())) { + logger.warn({ scheduleValue: data.schedule_value }, 'Invalid timestamp'); + return { ok: false }; + } + return { ok: true, nextRun: date.toISOString() }; + } + + return { ok: true, nextRun: null }; +} diff --git a/src/ipc-task-processor/tasks.ts b/src/ipc-task-processor/tasks.ts new file mode 100644 index 0000000..bf08ada --- /dev/null +++ b/src/ipc-task-processor/tasks.ts @@ -0,0 +1,138 @@ +import { CronExpressionParser } from 'cron-parser'; + +import { getTaskById, updateTask } from '../db.js'; +import { TIMEZONE } from '../config.js'; +import { logger } from '../logger.js'; +import type { TaskIpcPayload } from '../ipc-types.js'; + +type ScheduleType = 'cron' | 'interval' | 'once'; +type TaskUpdates = Parameters[1]; +type NextRunResolution = { abort: true } | { abort: false; nextRun?: string }; + +export function handleTaskStateMutation( + data: TaskIpcPayload, + sourceGroup: string, + isMain: boolean, + config: { + action: () => void; + successMessage: string; + unauthorizedMessage: string; + }, +): void { + if (!data.taskId) return; + + const task = getTaskById(data.taskId); + if (task && (isMain || task.group_folder === sourceGroup)) { + config.action(); + logger.info({ taskId: data.taskId, sourceGroup }, config.successMessage); + } else { + logger.warn( + { taskId: data.taskId, sourceGroup }, + config.unauthorizedMessage, + ); + } +} + +export function handleUpdateTask( + data: TaskIpcPayload, + sourceGroup: string, + isMain: boolean, +): void { + if (!data.taskId) return; + + const task = getTaskById(data.taskId); + if (!task) { + logger.warn( + { taskId: data.taskId, sourceGroup }, + 'Task not found for update', + ); + return; + } + if (!isMain && task.group_folder !== sourceGroup) { + logger.warn( + { taskId: data.taskId, sourceGroup }, + 'Unauthorized task update attempt', + ); + return; + } + + const updates = buildTaskUpdates(data, { + scheduleType: task.schedule_type, + scheduleValue: task.schedule_value, + }); + if (!updates) return; + + updateTask(data.taskId, updates); + logger.info( + { taskId: data.taskId, sourceGroup, updates }, + 'Task updated via IPC', + ); +} + +function buildTaskUpdates( + data: TaskIpcPayload, + existing: { scheduleType: string; scheduleValue: string }, +): TaskUpdates | null { + const updates: TaskUpdates = {}; + if (data.prompt !== undefined) updates.prompt = data.prompt; + if (data.schedule_type !== undefined) { + updates.schedule_type = data.schedule_type as ScheduleType; + } + if (data.schedule_value !== undefined) { + updates.schedule_value = data.schedule_value; + } + + if (data.schedule_type || data.schedule_value) { + const updatedScheduleType = updates.schedule_type ?? existing.scheduleType; + const updatedScheduleValue = + updates.schedule_value ?? existing.scheduleValue; + const nextRunResolution = resolveUpdatedNextRun({ + scheduleType: updatedScheduleType, + scheduleValue: updatedScheduleValue, + taskId: data.taskId, + }); + if (nextRunResolution.abort) return null; + if (nextRunResolution.nextRun !== undefined) { + updates.next_run = nextRunResolution.nextRun; + } + } + + return updates; +} + +function resolveUpdatedNextRun(args: { + scheduleType: string | undefined; + scheduleValue: string | null | undefined; + taskId: string | undefined; +}): NextRunResolution { + if (args.scheduleType === 'cron') { + try { + const interval = CronExpressionParser.parse(args.scheduleValue ?? '', { + tz: TIMEZONE, + }); + return { + abort: false, + nextRun: interval.next().toISOString() ?? undefined, + }; + } catch { + logger.warn( + { taskId: args.taskId, value: args.scheduleValue }, + 'Invalid cron in task update', + ); + return { abort: true }; + } + } + + if (args.scheduleType === 'interval') { + const ms = parseInt(args.scheduleValue ?? '', 10); + return { + abort: false, + nextRun: + !isNaN(ms) && ms > 0 + ? new Date(Date.now() + ms).toISOString() + : undefined, + }; + } + + return { abort: false }; +}