From 8b5497d115a485219036a27218c525f48f024db3 Mon Sep 17 00:00:00 2001 From: "[._.]/ Adam Eivy" Date: Sun, 6 Sep 2026 14:39:07 +0000 Subject: [PATCH] fix: continue scheduled perpetual drains after their cron slot --- server/services/cos.js | 7 ++++++- server/services/cos.test.js | 2 +- server/services/cosTaskGenerator.js | 7 +++++-- server/services/taskSchedule.js | 18 ++++++++++------ server/services/taskSchedule.test.js | 31 ++++++++++++++++++++++++++++ 5 files changed, 55 insertions(+), 10 deletions(-) diff --git a/server/services/cos.js b/server/services/cos.js index e05c2bf913..74c1d6d52a 100644 --- a/server/services/cos.js +++ b/server/services/cos.js @@ -1616,7 +1616,12 @@ async function refillPerpetualForCompletedAgent(agent) { // regenerates an identical first-line per app) is rejected as a duplicate of // the completing task and the drain stalls until the next scheduler tick. const cosTaskData = await getCosTasks(); - await queueEligibleImprovementTasks(state, cosTaskData, { ignoreTaskId: agent?.taskId, wakeAfterRecord: false }); + await queueEligibleImprovementTasks(state, cosTaskData, { + ignoreTaskId: agent?.taskId, + wakeAfterRecord: false, + // Continue only this completed drain: its cron slot already initiated it. + perpetualContinuation: { taskType: agentScheduledType(agent), appId: agent?.metadata?.taskApp || null } + }); // NOTE: the caller (the agent:completed handler) runs dequeueNextTask AFTER // this resolves, so the freshly-queued perpetual task is on the queue before // slots are filled. Do not dequeue here — that would re-introduce the ordering diff --git a/server/services/cos.test.js b/server/services/cos.test.js index a1042cf1fb..d1fb6b5268 100644 --- a/server/services/cos.test.js +++ b/server/services/cos.test.js @@ -1817,7 +1817,7 @@ describe('cos.js source — priority + capacity invariants', () => { expect( fnBody, 'queue path must constrain the pick to perpetual when on cooldown (perpetualOnly gated on cooldown)' - ).toMatch(/getNextTaskType\([^)]*\{\s*perpetualOnly:\s*onCooldown\s*\}/); + ).toMatch(/getNextTaskType\([^)]*\{\s*perpetualOnly:\s*onCooldown\s*[,}]/); }); it('generateManagedAppImprovementTaskForType defers updateAppActivity until after gates', () => { diff --git a/server/services/cosTaskGenerator.js b/server/services/cosTaskGenerator.js index b8634cf85f..9fbcce1948 100644 --- a/server/services/cosTaskGenerator.js +++ b/server/services/cosTaskGenerator.js @@ -1514,7 +1514,7 @@ export async function queueDueInstallWideImprovementTasks({ * Called during every evaluation to ensure system tasks are queued even when user tasks exist * Tasks are queued to COS-TASKS.md and will be picked up in Priority 2 */ -export async function queueEligibleImprovementTasks(state, cosTaskData, { ignoreTaskId = null, wakeAfterRecord = true } = {}) { +export async function queueEligibleImprovementTasks(state, cosTaskData, { ignoreTaskId = null, wakeAfterRecord = true, perpetualContinuation = null } = {}) { const taskSchedule = await import('./taskSchedule.js'); const { getDueTasks, getNextTaskType, recordExecution } = taskSchedule; @@ -1598,7 +1598,10 @@ export async function queueEligibleImprovementTasks(state, cosTaskData, { ignore // alone). When NOT on cooldown, the normal full-priority pick runs. const onCooldown = isAppActivityOnCooldown(appActivity, state.config.appReviewCooldownMs); - const nextTypeResult = await getNextTaskType(app.id, { perpetualOnly: onCooldown }).catch(() => null); + const nextTypeResult = await getNextTaskType(app.id, { + perpetualOnly: onCooldown, + continuingTaskType: perpetualContinuation?.appId === app.id ? perpetualContinuation.taskType : null + }).catch(() => null); if (!nextTypeResult) continue; const nextType = nextTypeResult.taskType; diff --git a/server/services/taskSchedule.js b/server/services/taskSchedule.js index 750c0b71d5..0f1ce8c455 100644 --- a/server/services/taskSchedule.js +++ b/server/services/taskSchedule.js @@ -790,9 +790,11 @@ async function checkRunAfterDeps(schedule, taskType, appId = null, featureEnable } /** - * Check if a task type should run for a specific app (or globally) + * Check if a task type should run for a specific app (or globally). + * Successful completion may continue its perpetual drain past the initiating + * cron slot; all eligibility and park gates still apply. */ -export async function shouldRunTask(taskType, appId = null, { featureEnabled = createFeatureGate() } = {}) { +export async function shouldRunTask(taskType, appId = null, { featureEnabled = createFeatureGate(), continuePerpetual = false } = {}) { if (appId && requiresInstallWideTarget(taskType)) { return { shouldRun: false, reason: 'requires-install-wide-target' }; } @@ -906,6 +908,10 @@ export async function shouldRunTask(taskType, appId = null, { featureEnabled = c if (isPerpetual) { const parked = perpetualParkResult(); if (parked) { result = parked; break; } + if (continuePerpetual) { + result = { shouldRun: true, reason: 'perpetual-drain' }; + break; + } // Unparked: the cron evaluation below decides whether to INITIATE a // drain. Once one is running, the completion-refill lane keeps it going // back-to-back regardless of subsequent ticks. @@ -1015,7 +1021,7 @@ export async function shouldRunTask(taskType, appId = null, { featureEnabled = c /** * Get all enabled task types that are due to run (optionally for a specific app) */ -export async function getDueTasks(appId = null) { +export async function getDueTasks(appId = null, { continuingTaskType = null } = {}) { const schedule = await loadSchedule(); const due = []; const featureEnabled = createFeatureGate(); @@ -1023,7 +1029,7 @@ export async function getDueTasks(appId = null) { for (const [taskType, interval] of Object.entries(schedule.tasks)) { if (!interval.enabled) continue; - const check = await shouldRunTask(taskType, appId, { featureEnabled }); + const check = await shouldRunTask(taskType, appId, { featureEnabled, continuePerpetual: taskType === continuingTaskType }); if (check.shouldRun) { due.push({ taskType, reason: check.reason, interval }); } @@ -1035,8 +1041,8 @@ export async function getDueTasks(appId = null) { /** * Get the next task type to run (optionally for a specific app) */ -export async function getNextTaskType(appId = null, { perpetualOnly = false } = {}) { - const dueTasks = await getDueTasks(appId); +export async function getNextTaskType(appId = null, { perpetualOnly = false, continuingTaskType = null } = {}) { + const dueTasks = await getDueTasks(appId, { continuingTaskType }); // `perpetualOnly` constrains the pick to a due perpetual (drain-until-done) // task, skipping every other schedule type. Callers set this when the app is diff --git a/server/services/taskSchedule.test.js b/server/services/taskSchedule.test.js index bed5e924d7..7f436812b2 100644 --- a/server/services/taskSchedule.test.js +++ b/server/services/taskSchedule.test.js @@ -2368,6 +2368,37 @@ describe('taskSchedule', () => { expect(await getNextTaskType()).toBeNull() }) + it('continues only the completed cron drain after its initiating slot is consumed', async () => { + cronNotDueYet() + mockSchedule({ + tasks: { + ...PAUSED_SHIPPED_DRAINS, + 'claim-issue': { type: 'cron', cronExpression: '0 7 * * *', perpetual: true, enabled: true }, + security: { type: 'cron', cronExpression: '0 7 * * *', enabled: true } + }, + executions: { 'task:claim-issue': { lastRun: new Date().toISOString(), count: 1, perApp: {} } } + }) + expect(await getNextTaskType()).toBeNull() + expect(await getNextTaskType(null, { continuingTaskType: 'security' })).toBeNull() + expect(await getNextTaskType(null, { continuingTaskType: 'claim-issue', perpetualOnly: true })) + .toEqual({ taskType: 'claim-issue', reason: 'perpetual-drain' }) + }) + + it.each(['parkedUntil', 'failureParkedAt'])('keeps continuation behind %s', async (field) => { + cronNotDueYet() + mockSchedule({ + tasks: { + ...PAUSED_SHIPPED_DRAINS, + 'claim-issue': { type: 'cron', cronExpression: '0 7 * * *', perpetual: true, enabled: true } + }, + executions: { 'task:claim-issue': { + lastRun: new Date().toISOString(), count: 1, perApp: {}, + [field]: new Date(Date.now() + 3600000).toISOString() + } } + }) + expect(await getNextTaskType(null, { continuingTaskType: 'claim-issue' })).toBeNull() + }) + it('an ELAPSED park makes a cron+perpetual task due immediately, without waiting for the next slot', async () => { cronNotDueYet() const past = new Date(Date.now() - 60 * 1000).toISOString()