fix: Codex round 18 — balance-disable idempotency, cluster reconcile, DB-time advance, disable-reason, resume overlap

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

@@ -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.

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,
 };

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;
           }

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<FireResult> {
   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.

packages/api/src/schedules/service.ts

@@ -76,6 +76,7 @@ export interface SchedulesService {
     limits: ScheduleLimits,
   ) => Promise<FireResult | null>;
   recordScheduleOutcome: (input: RecordScheduleOutcomeInput) => Promise<boolean>;
+  hasActiveScheduledRun: (scheduleId: string) => Promise<boolean>;
   markScheduleRunActive: (scheduleId: string, scheduledFor: string | Date) => Promise<void>;
   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<boolean> {
+    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,
   };

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<ISchedule>();
-    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<ISchedule>());
+    if (current?.enabled && current.balanceSkipCount >= balanceSkipDisableThreshold) {
       await disableSchedule(data.scheduleId, 'insufficient_balance');
     }
   }