claude-bot/EJClaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: split ipc task processor handlers

Browse Source
This commit is contained in:
ejclaw
2026-04-29 13:42:36 +09:00
parent 8855f6f07d
commit f286b9f99e
1 changed files with 576 additions and 527 deletions
Download Patch File Download Diff File Expand all files Collapse all files

499
src/ipc-task-processor.ts
View File

@@ -22,6 +22,11 @@ import {
} from './task-watch-status.js';
import type { IpcDeps, TaskIpcPayload } from './ipc-types.js';
type RoomBindings = ReturnType<IpcDeps['roomBindings']>;
type ScheduleType = 'cron' | 'interval' | 'once';
type TaskUpdates = Parameters<typeof updateTask>[1];
type MemorySourceKind = Parameters<typeof rememberMemory>[0]['sourceKind'];
export async function processTaskIpc(
data: TaskIpcPayload,
sourceGroup: string,
@@ -32,25 +37,85 @@ export async function processTaskIpc(
switch (data.type) {
case 'schedule_task':
handleScheduleTask(data, sourceGroup, isMain, deps, roomBindings);
break;
case 'pause_task':
handleTaskStateMutation(data, sourceGroup, isMain, {
action: () => updateTask(data.taskId!, { status: 'paused' }),
successMessage: 'Task paused via IPC',
unauthorizedMessage: 'Unauthorized task pause attempt',
});
break;
case 'resume_task':
handleTaskStateMutation(data, sourceGroup, isMain, {
action: () => updateTask(data.taskId!, { status: 'active' }),
successMessage: 'Task resumed via IPC',
unauthorizedMessage: 'Unauthorized task resume attempt',
});
break;
case 'cancel_task':
handleTaskStateMutation(data, sourceGroup, isMain, {
action: () => deleteTask(data.taskId!),
successMessage: 'Task cancelled via IPC',
unauthorizedMessage: 'Unauthorized task cancel attempt',
});
break;
case 'host_evidence_request':
await handleHostEvidenceRequest(data, sourceGroup, isMain, deps);
break;
case 'update_task':
handleUpdateTask(data, sourceGroup, isMain);
break;
case 'refresh_groups':
await handleRefreshGroups(sourceGroup, isMain, deps);
break;
case 'assign_room':
handleAssignRoom(data, sourceGroup, isMain, deps);
break;
case 'persist_memory':
handlePersistMemory(data, sourceGroup);
break;
default:
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
!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,
);
Object.values(roomBindings).find((group) => group.folder === targetJid);
if (!targetGroupEntry) {
logger.warn(
{ targetJid },
'Cannot schedule task: target group not registered',
);
break;
return;
}
const targetFolder = targetGroupEntry.folder;
@@ -66,7 +131,7 @@ export async function processTaskIpc(
{ targetJid, targetFolder },
'Cannot resolve scheduled task target JID from folder',
);
break;
return;
}
if (!isMain && targetFolder !== sourceGroup) {
@@ -74,48 +139,12 @@ export async function processTaskIpc(
{ sourceGroup, targetFolder },
'Unauthorized schedule_task attempt blocked',
);
break;
return;
}
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();
}
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(
@@ -133,7 +162,7 @@ export async function processTaskIpc(
},
'Duplicate CI watcher skipped — another agent already watches this run',
);
break;
return;
}
}
@@ -158,7 +187,7 @@ export async function processTaskIpc(
schedule_type: scheduleType,
schedule_value: data.schedule_value,
context_mode: contextMode,
next_run: nextRun,
next_run: nextRunResult.nextRun,
status: 'active',
created_at: new Date().toISOString(),
});
@@ -172,136 +201,100 @@ export async function processTaskIpc(
},
'Task created via IPC',
);
if (nextRun && new Date(nextRun).getTime() <= Date.now()) {
if (
nextRunResult.nextRun &&
new Date(nextRunResult.nextRun).getTime() <= Date.now()
) {
deps.nudgeScheduler?.();
}
}
break;
case 'pause_task':
if (data.taskId) {
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)) {
updateTask(data.taskId, { status: 'paused' });
logger.info(
{ taskId: data.taskId, sourceGroup },
'Task paused via IPC',
);
config.action();
logger.info({ taskId: data.taskId, sourceGroup }, config.successMessage);
} else {
logger.warn(
{ taskId: data.taskId, sourceGroup },
'Unauthorized task pause attempt',
config.unauthorizedMessage,
);
}
}
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 'host_evidence_request':
async function handleHostEvidenceRequest(
data: TaskIpcPayload,
sourceGroup: string,
isMain: boolean,
deps: IpcDeps,
): Promise<void> {
if (!data.requestId) {
logger.warn(
{ sourceGroup },
'Ignoring host_evidence_request without requestId',
);
break;
return;
}
if (data.action === 'ejclaw_room_runtime') {
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',
);
break;
}
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',
);
break;
}
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',
);
break;
handleRoomRuntimeReportRequest(data, sourceGroup, isMain, deps);
return;
}
if (!isHostEvidenceAction(data.action)) {
@@ -319,10 +312,9 @@ export async function processTaskIpc(
{ sourceGroup, requestId: data.requestId, action: data.action },
'Rejected unsupported host evidence action',
);
break;
return;
}
{
const result = await runHostEvidenceRequest({
requestId: data.requestId,
action: data.action,
@@ -346,48 +338,115 @@ export async function processTaskIpc(
'Processed host evidence request via IPC',
);
}
break;
case 'update_task':
if (data.taskId) {
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',
);
break;
return;
}
if (!isMain && task.group_folder !== sourceGroup) {
logger.warn(
{ taskId: data.taskId, sourceGroup },
'Unauthorized task update attempt',
);
break;
return;
}
const updates: Parameters<typeof updateTask>[1] = {};
const updates: TaskUpdates = {};
if (data.prompt !== undefined) updates.prompt = data.prompt;
if (data.schedule_type !== undefined) {
updates.schedule_type = data.schedule_type as
| 'cron'
| 'interval'
| 'once';
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,
};
const updatedTask = { ...task, ...updates };
if (updatedTask.schedule_type === 'cron') {
try {
const interval = CronExpressionParser.parse(
updatedTask.schedule_value,
{ tz: TIMEZONE },
{
tz: TIMEZONE,
},
);
updates.next_run = interval.next().toISOString();
} catch {
@@ -395,7 +454,7 @@ export async function processTaskIpc(
{ taskId: data.taskId, value: updatedTask.schedule_value },
'Invalid cron in task update',
);
break;
return;
}
} else if (updatedTask.schedule_type === 'interval') {
const ms = parseInt(updatedTask.schedule_value, 10);
@@ -411,40 +470,49 @@ export async function processTaskIpc(
'Task updated via IPC',
);
}
break;
case 'refresh_groups':
if (isMain) {
logger.info(
{ sourceGroup },
'Group metadata refresh requested via IPC',
);
async function handleRefreshGroups(
sourceGroup: string,
isMain: boolean,
deps: IpcDeps,
): Promise<void> {
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);
} else {
logger.warn(
{ sourceGroup },
'Unauthorized refresh_groups attempt blocked',
);
}
break;
case 'assign_room':
function handleAssignRoom(
data: TaskIpcPayload,
sourceGroup: string,
isMain: boolean,
deps: IpcDeps,
): void {
if (!isMain) {
logger.warn(
{ sourceGroup, type: data.type },
`Unauthorized ${data.type} attempt blocked`,
);
break;
return;
}
if (!data.jid || !data.name) {
logger.warn(
{ data },
`Invalid ${data.type} request - missing required fields`,
);
return;
}
if (data.jid && data.name) {
if (data.folder && !isValidGroupFolder(data.folder)) {
logger.warn(
{ sourceGroup, folder: data.folder },
`Invalid ${data.type} request - unsafe folder name`,
);
break;
return;
}
if (
data.room_mode !== undefined &&
@@ -455,7 +523,7 @@ export async function processTaskIpc(
{ sourceGroup, roomMode: data.room_mode },
'Invalid assign_room request - unknown room_mode',
);
break;
return;
}
if (
data.owner_agent_type !== undefined &&
@@ -466,7 +534,7 @@ export async function processTaskIpc(
{ sourceGroup, ownerAgentType: data.owner_agent_type },
'Invalid assign_room request - unknown owner_agent_type',
);
break;
return;
}
if (
data.reviewer_agent_type !== undefined &&
@@ -477,7 +545,7 @@ export async function processTaskIpc(
{ sourceGroup, reviewerAgentType: data.reviewer_agent_type },
'Invalid assign_room request - unknown reviewer_agent_type',
);
break;
return;
}
if (
data.arbiter_agent_type !== undefined &&
@@ -489,7 +557,7 @@ export async function processTaskIpc(
{ sourceGroup, arbiterAgentType: data.arbiter_agent_type },
'Invalid assign_room request - unknown arbiter_agent_type',
);
break;
return;
}
deps.assignRoom(data.jid, {
@@ -502,15 +570,9 @@ export async function processTaskIpc(
isMain: data.isMain,
workDir: data.workDir,
});
} else {
logger.warn(
{ data },
`Invalid ${data.type} request - missing required fields`,
);
}
break;
case 'persist_memory': {
function handlePersistMemory(data: TaskIpcPayload, sourceGroup: string): void {
if (
data.scopeKind !== 'room' ||
typeof data.scopeKey !== 'string' ||
@@ -520,7 +582,7 @@ export async function processTaskIpc(
{ sourceGroup, data },
'Invalid persist_memory request - missing required fields',
);
break;
return;
}
const expectedScopeKey = `room:${sourceGroup}`;
@@ -529,7 +591,7 @@ export async function processTaskIpc(
{ sourceGroup, scopeKey: data.scopeKey, expectedScopeKey },
'Unauthorized persist_memory attempt blocked',
);
break;
return;
}
if (
@@ -543,7 +605,7 @@ export async function processTaskIpc(
{ sourceGroup, sourceKind: data.source_kind },
'Invalid persist_memory request - unknown source_kind',
);
break;
return;
}
if (
@@ -554,7 +616,7 @@ export async function processTaskIpc(
{ sourceGroup, keywords: data.keywords },
'Invalid persist_memory request - keywords must be strings',
);
break;
return;
}
rememberMemory({
@@ -562,15 +624,8 @@ export async function processTaskIpc(
scopeKey: data.scopeKey,
content: data.content,
keywords: data.keywords,
memoryKind:
typeof data.memory_kind === 'string' ? data.memory_kind : null,
sourceKind:
(data.source_kind as
| 'compact'
| 'explicit'
| 'import'
| 'system'
| undefined) ?? 'compact',
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(
@@ -581,10 +636,4 @@ export async function processTaskIpc(
},
'Memory persisted via IPC',
);
break;
}
default:
logger.warn({ type: data.type }, 'Unknown IPC task type');
}
}