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

refactor: separate scheduler tick execution path

Browse Source
This commit is contained in:
Eyejoker
2026-03-31 07:26:21 +09:00
parent ca48e94291
commit aad1e93df5
2 changed files with 136 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
src/task-scheduler.test.ts
View File

@@ -142,6 +142,7 @@ import {
isWatchCiTask,
nudgeSchedulerLoop,
renderWatchCiStatusMessage,
runSchedulerTickOnce,
startSchedulerLoop,
} from './task-scheduler.js';
@@ -283,6 +284,48 @@ describe('task scheduler', () => {
);
});
it('can execute one scheduler tick without starting the timer loop', async () => {
const dueAt = new Date(Date.now() - 60_000).toISOString();
createTask({
id: 'task-single-tick',
group_folder: 'shared-group',
chat_jid: 'shared@g.us',
agent_type: 'codex',
prompt: 'single tick task',
schedule_type: 'once',
schedule_value: dueAt,
context_mode: 'isolated',
next_run: dueAt,
status: 'active',
created_at: '2026-02-22T00:00:00.000Z',
});
const enqueueTask = vi.fn();
await runSchedulerTickOnce({
serviceAgentType: 'codex',
registeredGroups: () => ({
'shared@g.us': {
name: 'Shared',
folder: 'shared-group',
trigger: '@Codex',
added_at: '2026-02-22T00:00:00.000Z',
agentType: 'codex',
},
}),
getSessions: () => ({}),
queue: { enqueueTask } as any,
onProcess: () => {},
sendMessage: async () => {},
});
expect(enqueueTask).toHaveBeenCalledTimes(1);
expect(enqueueTask.mock.calls[0][0]).toBe(
'shared@g.us::task:task-single-tick',
);
expect(enqueueTask.mock.calls[0][1]).toBe('task-single-tick');
});
it('keeps group-context tasks on the chat queue key', async () => {
const dueAt = new Date(Date.now() - 60_000).toISOString();
createTask({
@@ -1284,6 +1327,44 @@ Check the run.
expect(runAgentProcessMock).not.toHaveBeenCalled();
});
it('deletes expired tasks during a direct scheduler tick before enqueueing', async () => {
const enqueueTask = vi.fn();
createTask({
id: 'task-watch-expired-direct',
group_folder: 'shared-group',
chat_jid: 'shared@g.us',
agent_type: 'codex',
max_duration_ms: 60_000,
prompt: 'expired direct tick task',
schedule_type: 'interval',
schedule_value: '60000',
context_mode: 'isolated',
next_run: new Date(Date.now() - 60_000).toISOString(),
status: 'active',
created_at: '2026-02-22T00:00:00.000Z',
});
await runSchedulerTickOnce({
serviceAgentType: 'codex',
registeredGroups: () => ({
'shared@g.us': {
name: 'Shared',
folder: 'shared-group',
trigger: '@Codex',
added_at: '2026-02-22T00:00:00.000Z',
agentType: 'codex',
},
}),
getSessions: () => ({ 'shared-group': 'session-123' }),
queue: { enqueueTask } as any,
onProcess: () => {},
sendMessage: async () => {},
});
expect(getTaskById('task-watch-expired-direct')).toBeUndefined();
expect(enqueueTask).not.toHaveBeenCalled();
});
it('isolates both IPC and session state for isolated tasks', async () => {
const dueAt = new Date(Date.now() - 60_000).toISOString();
createTask({

104
src/task-scheduler.ts
View File

@@ -857,6 +857,60 @@ async function runGithubCiTask(
);
}
/**
* Execute one scheduler tick without timer/in-flight coordination.
* This keeps the scheduler loop focused on timing while tests and debugging
* can exercise the due-task path directly.
*/
export async function runSchedulerTickOnce(
deps: SchedulerDependencies,
): Promise<void> {
// Unified service: process all agent types, not just the service default.
const nowMs = Date.now();
const activeTasks = getAllTasks().filter((task) => task.status === 'active');
for (const task of activeTasks) {
const currentTask = getTaskById(task.id);
if (!currentTask || currentTask.status !== 'active') {
continue;
}
if (!hasTaskExceededMaxDuration(currentTask, nowMs)) {
continue;
}
deleteTask(currentTask.id);
logger.warn(
{
taskId: currentTask.id,
groupFolder: currentTask.group_folder,
maxDurationMs: currentTask.max_duration_ms,
createdAt: currentTask.created_at,
},
'Deleted task that exceeded max duration',
);
}
const dueTasks = getDueTasks();
if (dueTasks.length > 0) {
logger.info({ count: dueTasks.length }, 'Found due tasks');
}
for (const task of dueTasks) {
// Re-check task status in case it was paused/cancelled
const currentTask = getTaskById(task.id);
if (!currentTask || currentTask.status !== 'active') {
continue;
}
deps.queue.enqueueTask(getTaskQueueJid(currentTask), currentTask.id, () =>
isGitHubCiTask(currentTask)
? runGithubCiTask(currentTask, deps)
: runTask(currentTask, deps),
);
}
}
let schedulerRunning = false;
let schedulerTimer: ReturnType<typeof setTimeout> | null = null;
let schedulerLoopFn: (() => Promise<void>) | null = null;
@@ -899,55 +953,7 @@ export function startSchedulerLoop(deps: SchedulerDependencies): void {
schedulerTickInFlight = true;
try {
// Unified service: process all agent types, not just the service default.
const nowMs = Date.now();
const activeTasks = getAllTasks().filter(
(task) => task.status === 'active',
);
for (const task of activeTasks) {
const currentTask = getTaskById(task.id);
if (!currentTask || currentTask.status !== 'active') {
continue;
}
if (!hasTaskExceededMaxDuration(currentTask, nowMs)) {
continue;
}
deleteTask(currentTask.id);
logger.warn(
{
taskId: currentTask.id,
groupFolder: currentTask.group_folder,
maxDurationMs: currentTask.max_duration_ms,
createdAt: currentTask.created_at,
},
'Deleted task that exceeded max duration',
);
}
const dueTasks = getDueTasks();
if (dueTasks.length > 0) {
logger.info({ count: dueTasks.length }, 'Found due tasks');
}
for (const task of dueTasks) {
// Re-check task status in case it was paused/cancelled
const currentTask = getTaskById(task.id);
if (!currentTask || currentTask.status !== 'active') {
continue;
}
deps.queue.enqueueTask(
getTaskQueueJid(currentTask),
currentTask.id,
() =>
isGitHubCiTask(currentTask)
? runGithubCiTask(currentTask, deps)
: runTask(currentTask, deps),
);
}
await runSchedulerTickOnce(deps);
} catch (err) {
logger.error({ err }, 'Error in scheduler loop');
} finally {