fix: Codex round 17 — DB-time misfire/manual-lease, advance guard, cluster reconcile

1. engine.ts: derive the misfire cutoff from the claimed lease (leaseUntil -
   LEASE_MS = Mongo's claim time) instead of the worker clock, so a skewed replica
   can't drop a just-claimed occurrence as stale.
2. schedule.ts: acquireManualRunLease uses the same $$NOW CAS as claimDueSchedule,
   so a clock-ahead run-now can't read a Mongo-written lease as expired and start a
   second run.
4. advanceSchedule predicates its nextRunAt write on the claimed occurrence
   (expectedNextRunAt), so a concurrent owner edit that recomputed nextRunAt isn't
   clobbered by a stale-cadence value; threaded through fire.ts + engine.ts.
5. Reconciliation's job-status pass is gated on isJobStoreShared: a clustered
   backend with private per-worker in-memory stores (experimental.js passes
   USE_REDIS) skips it so a non-owning worker can't mislabel a peer's live run;
   the standard single-process backend stays fully reconciling.

(#3 architecture move to packages/api flagged on the PR for the author's call.)

api/server/experimental.js

@@ -496,7 +496,10 @@ if (cluster.isMaster) {
         await checkMigrations();
         // Arm the scheduler in each worker: the Mongo lease-claim CAS guarantees
         // exactly one worker fires each due schedule, so multi-worker arming is safe.
-        await initializeScheduleEngine();
+        // The job store is only SHARED across these workers when Redis-backed; with
+        // private per-worker in-memory stores a worker can't observe a peer's jobs,
+        // so mark it unshared to skip job-status reconciliation it would misread.
+        await initializeScheduleEngine({ isJobStoreShared: isEnabled(process.env.USE_REDIS) });
       } catch (initErr) {
         logger.error(`Worker ${process.pid} post-listen initialization failed:`, initErr);
         process.exit(1);

api/server/services/Schedules/index.js

@@ -50,6 +50,12 @@ async function getLimits(user) {
 
 const MANUAL_RUN_LEASE_MS = 5 * 60 * 1000;
 
+// Whether every engine replica observes the same jobs. The standard backend is a
+// single process (its one engine sees all its jobs), so it defaults true and keeps
+// full reconciliation; a clustered backend with private in-memory stores passes
+// false unless Redis-backed (see initializeScheduleEngine / experimental.js).
+let jobStoreShared = true;
+
 /**
  * Whether a refill would top up this zero-credit balance record right now,
  * mirroring the chat balance check's auto-refill eligibility (record-based).
@@ -166,6 +172,7 @@ const engineDeps = {
     const { GenerationJobManager } = require('@librechat/api');
     await GenerationJobManager.getJobStore()?.deleteJob(conversationId);
   },
+  isJobStoreShared: () => jobStoreShared,
   // Counted in system scope so the cap is GLOBAL — a per-owner (tenant-scoped)
   // count would let multiple tenants collectively exceed fireConcurrency.
   countActiveRunsGlobal: () => runAsSystem(() => methods.countActiveRuns()),
@@ -174,10 +181,15 @@ const engineDeps = {
 /** @type {ReturnType<typeof startScheduleEngine> | undefined} */
 let engine;
 
-async function initializeScheduleEngine() {
+async function initializeScheduleEngine(options) {
   if (engine != null) {
     return engine;
   }
+  // A clustered backend passes isJobStoreShared=false (unless Redis-backed) so the
+  // reconciler skips job-status checks it can't trust across workers.
+  if (options?.isJobStoreShared != null) {
+    jobStoreShared = options.isJobStoreShared;
+  }
   // Explicitly build the Schedule/ScheduleRun indexes first — the unique
   // idempotency index and TTL retention index would otherwise never exist when
   // MONGO_AUTO_INDEX is disabled (the production default). If this fails the

packages/api/src/schedules/engine.ts

@@ -36,12 +36,20 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine {
   async function reconcile() {
     try {
       const limits = await deps.getLimits();
-      const runs = await runAsSystem(() =>
-        deps.methods.getRunsForReconciliation(
-          new Date(Date.now() - RECONCILE_MIN_RUN_AGE_MS),
-          RECONCILE_BATCH,
-        ),
-      );
+      // 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,
+            ),
+          )
+        : [];
       await runAsSystem(async () => {
         for (const run of runs) {
           const jobStatus = run.conversationId ? await deps.getJobStatus(run.conversationId) : null;
@@ -174,10 +182,15 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine {
           return true;
         }
         const scheduledFor = schedule.nextRunAt ?? new Date();
+        // Use DB time, not this worker's clock, for the misfire cutoff: the claim
+        // wrote leaseUntil = $$NOW + LEASE_MS, so leaseUntil - LEASE_MS is Mongo's
+        // "now" at claim. A skewed worker would otherwise treat a just-claimed
+        // occurrence as stale and drop it (advance without firing).
+        const dbNow = schedule.leaseUntil ? schedule.leaseUntil.getTime() - LEASE_MS : Date.now();
         // Misfire skip-forward: an occurrence overdue past the grace window (the
         // engine was down/paused) is advanced to the next FUTURE occurrence
         // without firing, so a restart doesn't launch stale or bursty chats.
-        if (Date.now() - scheduledFor.getTime() > MISFIRE_GRACE_MS) {
+        if (dbNow - scheduledFor.getTime() > MISFIRE_GRACE_MS) {
           const next = computeNextRunAt({
             cadence: schedule.cadence,
             timezone: schedule.timezone,
@@ -188,7 +201,9 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine {
               .disableSchedule(schedule.id, 'invalid_schedule')
               .catch(() => undefined);
           }
-          await deps.methods.advanceSchedule(schedule.id, next).catch(() => undefined);
+          await deps.methods
+            .advanceSchedule(schedule.id, next, scheduledFor)
+            .catch(() => undefined);
           logger.info(`[schedules] skipped stale occurrence for ${schedule.id} (misfire grace)`);
           return false;
         }
@@ -212,7 +227,9 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine {
               .disableSchedule(schedule.id, 'invalid_schedule')
               .catch(() => undefined);
           }
-          await deps.methods.advanceSchedule(schedule.id, next).catch(() => undefined);
+          await deps.methods
+            .advanceSchedule(schedule.id, next, scheduledFor)
+            .catch(() => undefined);
         }
         return false;
       });

packages/api/src/schedules/fire.spec.ts

@@ -107,6 +107,7 @@ function makeDeps(
     runInTenantContext: (_user, fn) => fn(),
     getJobStatus: async () => null,
     clearReconciledJob: async () => undefined,
+    isJobStoreShared: () => true,
     countActiveRunsGlobal: async () => methods.countActiveRuns(),
     ...over,
   } as ScheduleEngineDeps;

packages/api/src/schedules/fire.ts

@@ -121,7 +121,9 @@ export async function fireSchedule(
   // only releases the lease it acquired for serialization.
   const advance = options?.manual
     ? () => methods.releaseLease(schedule.id)
-    : () => methods.advanceSchedule(schedule.id, nextRunAt);
+    : // Predicate the advance on the claimed occurrence so a concurrent owner edit
+      // that recomputed nextRunAt between claim and fire isn't clobbered.
+      () => methods.advanceSchedule(schedule.id, nextRunAt, scheduledFor);
 
   if (nextRunAt == null) {
     await methods.disableSchedule(schedule.id, 'invalid_schedule');

packages/api/src/schedules/types.ts

@@ -59,6 +59,14 @@ export interface ScheduleEngineDeps {
   runInTenantContext: <T>(user: ScheduleUserContext, fn: () => Promise<T>) => Promise<T>;
   /** Job-store status for a run's conversation, or null when the job is gone. */
   getJobStatus: (conversationId: string) => Promise<string | null>;
+  /**
+   * Whether every engine replica can observe the SAME jobs (Redis-backed, or a
+   * single process). When false — e.g. clustered workers each with a private
+   * in-memory store — a non-owning replica would read `jobStatus == null` for a
+   * peer's live run and wrongly interrupt it, so the job-status reconciliation is
+   * skipped and each run is finalized only by its owning replica's inline hooks.
+   */
+  isJobStoreShared: () => boolean;
   /**
    * Deletes a retained terminal job after the reconciler has finalized its run.
    * Gives `preserveForReconcile` jobs (kept without `completedAt` so the store's

packages/data-schemas/src/methods/schedule.ts

@@ -54,7 +54,11 @@ export type ScheduleMethods = {
     leaseMs: number,
   ) => Promise<boolean>;
   releaseLease: (id: string) => Promise<void>;
-  advanceSchedule: (id: string, nextRunAt: Date | null) => Promise<void>;
+  advanceSchedule: (
+    id: string,
+    nextRunAt: Date | null,
+    expectedNextRunAt?: Date | null,
+  ) => Promise<void>;
   disableSchedule: (id: string, reason: ScheduleDisabledReason) => Promise<void>;
   insertScheduleRun: (data: Partial<IScheduleRun>) => Promise<IScheduleRun | null>;
   setRunFireDetails: (
@@ -212,15 +216,17 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
     userId: string | Types.ObjectId,
     leaseMs: number,
   ): Promise<boolean> {
-    const now = new Date();
+    // Compare/expire the lease against Mongo's `$$NOW` (same CAS shape as
+    // claimDueSchedule), not this worker's clock: a skewed replica must not read a
+    // Mongo-written automatic-fire lease as expired early and start a second run.
     const row = await Schedule()
       .findOneAndUpdate(
         {
           id,
           user: userId,
-          $or: [{ leaseUntil: { $exists: false } }, { leaseUntil: { $lt: now } }],
+          $expr: { $lt: [{ $ifNull: ['$leaseUntil', new Date(0)] }, '$$NOW'] },
         },
-        { $set: { leaseUntil: new Date(now.getTime() + leaseMs), leaseBy: 'manual' } },
+        [{ $set: { leaseUntil: { $add: ['$$NOW', leaseMs] }, leaseBy: 'manual' } }],
         { new: true },
       )
       .lean<ISchedule>();
@@ -232,15 +238,27 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
     await Schedule().updateOne({ id }, { $unset: { leaseUntil: 1, leaseBy: 1 } });
   }
 
-  /** Advances past a fired (or skipped) occurrence and releases the lease. */
-  async function advanceSchedule(id: string, nextRunAt: Date | null): Promise<void> {
-    await Schedule().updateOne(
-      { id },
-      {
-        $set: { ...(nextRunAt ? { nextRunAt } : {}) },
-        $unset: { leaseUntil: 1, leaseBy: 1, ...(nextRunAt ? {} : { nextRunAt: 1 }) },
-      },
-    );
+  /**
+   * Advances past a fired (or skipped) occurrence and releases the lease. When
+   * `expectedNextRunAt` is given, the whole update is predicated on the schedule
+   * still sitting on the claimed occurrence — so a concurrent owner edit (or
+   * re-enable) that recomputed `nextRunAt` between the claim and here is NOT
+   * clobbered by a value derived from the stale cadence snapshot; the stale
+   * claimer's lease then simply expires on its own.
+   */
+  async function advanceSchedule(
+    id: string,
+    nextRunAt: Date | null,
+    expectedNextRunAt?: Date | null,
+  ): Promise<void> {
+    const filter: Record<string, unknown> = { id };
+    if (expectedNextRunAt !== undefined) {
+      filter.nextRunAt = expectedNextRunAt;
+    }
+    await Schedule().updateOne(filter, {
+      $set: { ...(nextRunAt ? { nextRunAt } : {}) },
+      $unset: { leaseUntil: 1, leaseBy: 1, ...(nextRunAt ? {} : { nextRunAt: 1 }) },
+    });
   }
 
   async function disableSchedule(id: string, reason: ScheduleDisabledReason): Promise<void> {