Merge pull request #176 from phj1081/codex/ipc-task-processor-quality
refactor: split ipc task processor handlers
This commit is contained in:
Eyejoker
@@ -109,11 +109,6 @@
|
|||||||
"maxFunctionLines": 314,
|
"maxFunctionLines": 314,
|
||||||
"owner": "main service entrypoint hotspot"
|
"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": {
|
"src/message-agent-executor-attempt-runner.ts": {
|
||||||
"maxFunctionLines": 275,
|
"maxFunctionLines": 275,
|
||||||
"maxComplexity": 47,
|
"maxComplexity": 47,
|
||||||
|
|||||||
@@ -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 ---
|
// --- cancel_task authorization ---
|
||||||
|
|
||||||
describe('cancel_task authorization', () => {
|
describe('cancel_task authorization', () => {
|
||||||
|
|||||||
@@ -1,41 +1,17 @@
|
|||||||
import { CronExpressionParser } from 'cron-parser';
|
import { deleteTask, updateTask } from './db.js';
|
||||||
|
|
||||||
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 { logger } from './logger.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 {
|
import {
|
||||||
DEFAULT_WATCH_CI_MAX_DURATION_MS,
|
handleAssignRoom,
|
||||||
isWatchCiTask,
|
handleRefreshGroups,
|
||||||
} from './task-watch-status.js';
|
} 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 { IpcDeps, TaskIpcPayload } from './ipc-types.js';
|
||||||
import type { PairedRoomRole } from './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'];
|
|
||||||
|
|
||||||
function normalizePairedRoomRole(
|
|
||||||
role: string | null | undefined,
|
|
||||||
): PairedRoomRole | undefined {
|
|
||||||
return role === 'owner' || role === 'reviewer' || role === 'arbiter'
|
|
||||||
? role
|
|
||||||
: undefined;
|
|
||||||
}
|
|
||||||
|
|
||||||
export async function processTaskIpc(
|
export async function processTaskIpc(
|
||||||
data: TaskIpcPayload,
|
data: TaskIpcPayload,
|
||||||
@@ -43,11 +19,9 @@ export async function processTaskIpc(
|
|||||||
isMain: boolean,
|
isMain: boolean,
|
||||||
deps: IpcDeps,
|
deps: IpcDeps,
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
const roomBindings = deps.roomBindings();
|
|
||||||
|
|
||||||
switch (data.type) {
|
switch (data.type) {
|
||||||
case 'schedule_task':
|
case 'schedule_task':
|
||||||
handleScheduleTask(data, sourceGroup, isMain, deps, roomBindings);
|
handleScheduleTask(data, sourceGroup, isMain, deps);
|
||||||
break;
|
break;
|
||||||
|
|
||||||
case 'pause_task':
|
case 'pause_task':
|
||||||
@@ -98,569 +72,3 @@ export async function processTaskIpc(
|
|||||||
logger.warn({ type: data.type }, 'Unknown IPC task type');
|
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<void> {
|
|
||||||
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<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);
|
|
||||||
}
|
|
||||||
|
|
||||||
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',
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|||||||
145
src/ipc-task-processor/host-evidence.ts
Normal file
145
src/ipc-task-processor/host-evidence.ts
Normal file
@@ -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<void> {
|
||||||
|
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',
|
||||||
|
);
|
||||||
|
}
|
||||||
96
src/ipc-task-processor/memory.ts
Normal file
96
src/ipc-task-processor/memory.ts
Normal file
@@ -0,0 +1,96 @@
|
|||||||
|
import { rememberMemory } from '../db.js';
|
||||||
|
import { logger } from '../logger.js';
|
||||||
|
import type { TaskIpcPayload } from '../ipc-types.js';
|
||||||
|
|
||||||
|
type MemorySourceKind = Parameters<typeof rememberMemory>[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'))
|
||||||
|
);
|
||||||
|
}
|
||||||
116
src/ipc-task-processor/rooms.ts
Normal file
116
src/ipc-task-processor/rooms.ts
Normal file
@@ -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<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);
|
||||||
|
}
|
||||||
|
|
||||||
|
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';
|
||||||
|
}
|
||||||
228
src/ipc-task-processor/schedule.ts
Normal file
228
src/ipc-task-processor/schedule.ts
Normal file
@@ -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<IpcDeps['roomBindings']>;
|
||||||
|
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 };
|
||||||
|
}
|
||||||
138
src/ipc-task-processor/tasks.ts
Normal file
138
src/ipc-task-processor/tasks.ts
Normal file
@@ -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<typeof updateTask>[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 };
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user