From 80b9197f239c1b265f008d900b3f1e130320c8aa Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Wed, 22 Jul 2026 04:23:04 -0400 Subject: [PATCH] =?UTF-8?q?fix:=20Codex=20round=2018=20=E2=80=94=20balance?= =?UTF-8?q?-disable=20idempotency,=20cluster=20reconcile,=20DB-time=20adva?= =?UTF-8?q?nce,=20disable-reason,=20resume=20overlap?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. schedule.ts: balance-skip auto-disable is now a policy re-evaluated outside the countedFor guard (re-read on a no-op'd guarded update), so a crash between the $inc and disableSchedule still disables on replay. 2. engine.ts: reconcile now processes runs THIS worker owns (non-null jobStatus) on every replica and only gates the jobStatus==null orphan/abandoned reaping on isJobStoreShared — so a non-Redis cluster still finalizes its own runs instead of dropping the whole pass. 3. fire.ts/engine.ts: computeNextRunAt for automatic fires uses the DB claim time (leaseUntil - LEASE_MS) via a dbNow option, so a clock-ahead worker can't skip valid future occurrences. 4. schedule.ts: the success-path disabledReason unset is predicated on enabled:true, so a stale older run succeeding after newer outcomes disabled the schedule doesn't wipe the reason. 5. service.ts/resume.js: before resuming a paused scheduled run, check for another active run and defer (409, before consuming the approval) so HITL resumes don't bypass per-schedule overlap serialization. --- api/server/controllers/agents/resume.js | 20 ++++++++- api/server/services/Schedules/index.js | 1 + packages/api/src/schedules/engine.ts | 43 +++++++++++-------- packages/api/src/schedules/fire.ts | 10 ++++- packages/api/src/schedules/service.ts | 12 ++++++ packages/data-schemas/src/methods/schedule.ts | 21 ++++++++- 6 files changed, 85 insertions(+), 22 deletions(-) diff --git a/api/server/controllers/agents/resume.js b/api/server/controllers/agents/resume.js index d5004ef128..5e9cb00d4f 100644 --- a/api/server/controllers/agents/resume.js +++ b/api/server/controllers/agents/resume.js @@ -23,7 +23,11 @@ const { getMCPRequestContext, cleanupMCPRequestContextForReq, } = require('~/server/services/MCPRequestContext'); -const { recordScheduleOutcome, markScheduleRunActive } = require('~/server/services/Schedules'); +const { + recordScheduleOutcome, + hasActiveScheduledRun, + markScheduleRunActive, +} = require('~/server/services/Schedules'); const { saveMessage, getConvo, getMessages } = require('~/models'); /** @@ -536,6 +540,20 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) return res.status(429).json({ error: 'Too many concurrent requests' }); } + // A scheduled fire's paused run must not resume OVER a newer occurrence that is + // already running: overlap checks only count `started`, so promoting this paused + // row would create two concurrent runs for one schedule. Defer the resume BEFORE + // claiming the approval (so it stays claimable) when another occurrence is active. + if (job.metadata?.scheduleId && (await hasActiveScheduledRun(job.metadata.scheduleId))) { + await decrementPendingRequest(userId); + logger.debug( + `[ResumeAgentController] Deferring scheduled resume; another run active: ${streamId}`, + ); + return res + .status(409) + .json({ error: 'Another run for this schedule is in progress; try again shortly' }); + } + // Atomically claim the resume. The single winner drives the run; a racing second // submit (double-click, two tabs) gets false and must not re-drive — that would // re-execute tools and double-bill. diff --git a/api/server/services/Schedules/index.js b/api/server/services/Schedules/index.js index eaabd83918..4bf03b33ba 100644 --- a/api/server/services/Schedules/index.js +++ b/api/server/services/Schedules/index.js @@ -24,6 +24,7 @@ module.exports = { engineDeps: service.engineDeps, fireScheduleNow: service.fireScheduleNow, recordScheduleOutcome: service.recordScheduleOutcome, + hasActiveScheduledRun: service.hasActiveScheduledRun, markScheduleRunActive: service.markScheduleRunActive, initializeScheduleEngine: service.initializeScheduleEngine, }; diff --git a/packages/api/src/schedules/engine.ts b/packages/api/src/schedules/engine.ts index 2e3ec88ae5..d16b3ee001 100644 --- a/packages/api/src/schedules/engine.ts +++ b/packages/api/src/schedules/engine.ts @@ -36,20 +36,19 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { async function reconcile() { try { const limits = await deps.getLimits(); - // The job-status pass can only be trusted when every replica sees the same - // jobs (Redis-backed or single-process). With private per-worker in-memory - // stores a non-owning worker reads jobStatus == null for a peer's live run - // and would wrongly interrupt it, so skip this pass there and let each run's - // owning worker finalize it inline. The bookkeeping-catch pass below is - // job-status-independent and always runs. - const runs = deps.isJobStoreShared() - ? await runAsSystem(() => - deps.methods.getRunsForReconciliation( - new Date(Date.now() - RECONCILE_MIN_RUN_AGE_MS), - RECONCILE_BATCH, - ), - ) - : []; + // When every replica shares the job store (Redis-backed or single-process), a + // jobStatus == null genuinely means the job is gone. With private per-worker + // in-memory stores it can instead mean the run's job lives on a PEER worker, so + // a non-owning worker must not treat null as orphaned. We still process runs + // this worker owns (non-null jobStatus = this worker's job) on every replica; + // only the jobStatus == null orphan/abandoned reaping below is gated on sharing. + const jobStoreShared = deps.isJobStoreShared(); + const runs = await runAsSystem(() => + deps.methods.getRunsForReconciliation( + new Date(Date.now() - RECONCILE_MIN_RUN_AGE_MS), + RECONCILE_BATCH, + ), + ); await runAsSystem(async () => { for (const run of runs) { const jobStatus = run.conversationId ? await deps.getJobStatus(run.conversationId) : null; @@ -110,12 +109,20 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { // run records its conversationId up front (fire.ts pre-generates it), so // getJobStatus above already liveness-checked it: a live long-running fire // reads as `running` and is left alone; only one whose job is genuinely - // gone reaches here, so the standard orphan cutoff applies. - if (run.status === 'started' && jobStatus == null && ageMs > ORPHAN_RUN_AGE_MS) { + // gone reaches here, so the standard orphan cutoff applies. Gated on + // jobStoreShared: with private per-worker stores a null could be a peer's + // live job, so a non-owning worker must not interrupt it (its owner does). + if ( + jobStoreShared && + run.status === 'started' && + jobStatus == null && + ageMs > ORPHAN_RUN_AGE_MS + ) { await finalize('interrupted'); continue; } if ( + jobStoreShared && run.status === 'requires_action' && jobStatus == null && ageMs > ABANDONED_PAUSE_AGE_MS @@ -208,7 +215,9 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { return false; } try { - const result = await fireSchedule(deps, schedule, limits, scheduledFor); + const result = await fireSchedule(deps, schedule, limits, scheduledFor, { + dbNow: new Date(dbNow), + }); if (result.fired) { fired += 1; } diff --git a/packages/api/src/schedules/fire.ts b/packages/api/src/schedules/fire.ts index 70ab2b1e29..8b06a033db 100644 --- a/packages/api/src/schedules/fire.ts +++ b/packages/api/src/schedules/fire.ts @@ -108,14 +108,20 @@ export async function fireSchedule( schedule: FireableSchedule, limits: ScheduleLimits, scheduledFor: Date, - options?: { manual?: boolean }, + options?: { manual?: boolean; dbNow?: Date }, ): Promise { const { methods } = deps; + // Compute the NEXT occurrence relative to DB time (the engine passes the claim + // time derived from leaseUntil), not this worker's clock: a clock-ahead worker + // would otherwise advance past valid future occurrences. Falls back to the + // process clock when no DB time is provided (e.g. manual run-now, which never + // reschedules and so ignores the result anyway). + const now = options?.dbNow ?? new Date(); const nextRunAt = computeNextRunAt({ cadence: schedule.cadence, timezone: schedule.timezone, scheduleId: schedule.id, - after: new Date(Math.max(Date.now(), scheduledFor.getTime())), + after: new Date(Math.max(now.getTime(), scheduledFor.getTime())), }); // A manual run-now must never reschedule the next automatic occurrence; it // only releases the lease it acquired for serialization. diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index 06a45e28e4..617d683e8f 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -76,6 +76,7 @@ export interface SchedulesService { limits: ScheduleLimits, ) => Promise; recordScheduleOutcome: (input: RecordScheduleOutcomeInput) => Promise; + hasActiveScheduledRun: (scheduleId: string) => Promise; markScheduleRunActive: (scheduleId: string, scheduledFor: string | Date) => Promise; initializeScheduleEngine: (options?: { isJobStoreShared?: boolean; @@ -347,6 +348,16 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer * concurrently. Best-effort: a failure just leaves it `requires_action` (the * terminal hook still records the outcome). */ + /** + * Whether another occurrence of this schedule is already running (`started`). + * A paused run being resumed is `requires_action`, not `started`, so a true here + * means a NEWER occurrence is active — the resume must defer/reject, or promoting + * this paused row to `started` would bypass per-schedule overlap serialization. + */ + async function hasActiveScheduledRun(scheduleId: string): Promise { + return methods.hasActiveRun(scheduleId); + } + async function markScheduleRunActive( scheduleId: string, scheduledFor: string | Date, @@ -371,6 +382,7 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer engineDeps, fireScheduleNow, recordScheduleOutcome, + hasActiveScheduledRun, markScheduleRunActive, initializeScheduleEngine, }; diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index 64ae906843..addb0eceb0 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -330,9 +330,20 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche $push: { countedFor: { $each: [params.scheduledFor], $slice: -COUNTED_FOR_WINDOW } }, ...(isSuccess ? { $inc: { runCount: 1 } } : {}), ...(isFailure ? { $inc: { failureCount: 1 } } : {}), - ...(isSuccess ? { $unset: { disabledReason: 1 } } : {}), }, ); + // A success clears a transient disable reason ONLY while the schedule is still + // enabled. An older run (e.g. a resumed pause) can succeed AFTER newer outcomes + // already auto-disabled the schedule — since `requires_action` runs don't block + // later occurrences — and must not wipe the reason that explains why it's off. + // Predicated on `enabled` separately so it can't leak into the count-guarded + // update above (which must run regardless of enabled state). + if (isSuccess) { + await Schedule().updateOne( + { id: params.scheduleId, enabled: true }, + { $unset: { disabledReason: 1 } }, + ); + } // Auto-disable is a POLICY re-evaluated on EVERY call (idempotent), NOT gated // on the count guard — so if a crash landed the $inc but not the disable, the // reconciler's replay still disables. Reads current state after the count. @@ -440,7 +451,13 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche { new: true }, ) .lean(); - if (schedule != null && schedule.balanceSkipCount >= balanceSkipDisableThreshold) { + // Auto-disable is a POLICY re-evaluated on EVERY call (idempotent), NOT gated on + // the count guard — mirroring applyTerminalBookkeeping. If a crash landed the + // $inc/$push but not the disable, the guarded update above no-ops to null on the + // replay, so re-read the current counter and still disable when at/over threshold. + const current = + schedule ?? (await Schedule().findOne({ id: data.scheduleId }).lean()); + if (current?.enabled && current.balanceSkipCount >= balanceSkipDisableThreshold) { await disableSchedule(data.scheduleId, 'insufficient_balance'); } }