From aad1e93df543f6dc21079849ae4a9cd83b096c5a Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Tue, 31 Mar 2026 07:26:21 +0900 Subject: [PATCH] refactor: separate scheduler tick execution path --- src/task-scheduler.test.ts | 81 +++++++++++++++++++++++++++++ src/task-scheduler.ts | 104 ++++++++++++++++++++----------------- 2 files changed, 136 insertions(+), 49 deletions(-) diff --git a/src/task-scheduler.test.ts b/src/task-scheduler.test.ts index 4045a3f..0e8faf5 100644 --- a/src/task-scheduler.test.ts +++ b/src/task-scheduler.test.ts @@ -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({ diff --git a/src/task-scheduler.ts b/src/task-scheduler.ts index 5cb972f..a7f2ed5 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -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 { + // 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 | null = null; let schedulerLoopFn: (() => Promise) | 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 {