fix(schedules): Wave C.1 — reconcile durable trigger delivery with the reservation

Threads r3803826192 (manual limiter) and r3804255924 (PII/moderation), plus the
independently-found long-Retry-After race. fireSchedule reserves a `started` run
and a global capacity slot BEFORE the durable trigger delivery reaches the chat
route, where an interactive limiter (manual Run Now), PII, or moderation can
reject it before any generation job exists — dead-lettering the delivery while
the run sat `started` until the 30-minute orphan sweep mislabeled it interrupted.
And a valid delivery deferred by Retry-After (up to 24h) could be orphan-settled
and have its capacity released, then fire anyway.

Translate durable delivery state into the schedule outcome:

- Store the deterministic trigger deliveryKey on the ScheduleRun reservation
  (computed from the envelope BEFORE enqueue, so an ambiguous commit still has it).
- Add a getTriggerDelivery engine dep (wired to the merged trigger service's
  getDelivery) that reads the durable delivery by key.
- Schedule reconciliation, for a jobless `started` run: staging/pending/leased →
  admission is live, never orphan; dead → record `error` from the durable
  lastError and release capacity promptly (no 30-minute wait), through the
  ordinary outcome/auto-disable path; succeeded or no record → the existing
  legacy orphan policy (interrupted only past the cutoff); a delivery lookup
  failure defers rather than orphaning a possibly-live delivery. Limiter/PII/
  moderation middleware writes no schedule state.

Adds reconcile state-mapping tests (dead/pending/leased/staging/succeeded/none/
lookup-failure) and a fire test that the reservation's deliveryKey equals the
enqueued delivery's idempotency key.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq

api/server/services/Schedules/index.js

@@ -13,7 +13,10 @@ function getService() {
   const mongoose = require('mongoose');
   const { createSchedulesService } = require('@librechat/api');
   const { getAppConfig } = require('~/server/services/Config/app');
-  const { enqueueAgentTrigger } = require('~/server/services/Agents/triggers');
+  const {
+    enqueueAgentTrigger,
+    getAgentTriggerDelivery,
+  } = require('~/server/services/Agents/triggers');
   const { resolveAgentFireAccess } = require('./access');
   const methods = require('~/models');
   const isUserDeleting = async (userId) => !(await methods.isAgentTriggerPrincipalActive(userId));
@@ -45,6 +48,9 @@ function getService() {
         { new: true },
       ).lean(),
     enqueueAgentTrigger,
+    // Reconciliation reads the durable delivery to tell a still-live admission (a
+    // deferred Retry-After) or a dead-letter apart from a genuinely orphaned run.
+    getTriggerDelivery: getAgentTriggerDelivery,
     resolveAgentFireAccess,
     // Reuse the merged trigger/deletion admission fence. A schedule fire and every
     // generic trigger now fail closed on the same durable principal state.

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

@@ -81,6 +81,7 @@ function makeDeps(
     hasScheduleAccess: async () => true,
     resolveFiles: async () => [],
     enqueueTrigger: jest.fn(async () => undefined),
+    getTriggerDelivery: async () => null,
     runInTenantContext: (_user, fn) => fn(),
     getJobStatus: async () => null,
     abortScheduledJob: async () => undefined,
@@ -224,6 +225,84 @@ describe('runTick error handling', () => {
   });
 });
 
+describe('reconciliation consults the durable trigger delivery', () => {
+  const YOUNG = () => new Date(Date.now() - 5 * 60_000); // past reconcile min, before orphan cutoff
+  const OLD = () => new Date(Date.now() - 60 * 60_000); // past the 30-minute orphan cutoff
+
+  function joblessRun(firedAt: Date, deliveryKey: string | undefined = 'dk1') {
+    return {
+      scheduleId: 's1',
+      scheduledFor: new Date('2026-01-01T00:00:00.000Z'),
+      user: 'u1',
+      status: 'started',
+      conversationId: 'conv-1',
+      firedAt,
+      ...(deliveryKey != null ? { deliveryKey } : {}),
+    };
+  }
+
+  async function runReconcile(
+    run: ReturnType<typeof joblessRun>,
+    getTriggerDelivery: ScheduleEngineDeps['getTriggerDelivery'],
+  ) {
+    const methods = makeMethods(makeClaimedSchedule());
+    (methods.getRunsForReconciliation as jest.Mock).mockResolvedValue([run]);
+    await tickOnce(makeDeps(methods, { getJobStatus: async () => null, getTriggerDelivery }));
+    return methods;
+  }
+
+  it('settles a DEAD delivery as error promptly, before the 30-minute orphan age', async () => {
+    const methods = await runReconcile(joblessRun(YOUNG()), async () => ({
+      status: 'dead',
+      lastError: 'rate limited',
+    }));
+    expect(methods.recordRunOutcome).toHaveBeenCalledWith(
+      expect.objectContaining({ scheduleId: 's1', status: 'error', error: 'rate limited' }),
+    );
+  });
+
+  it('does NOT orphan a PENDING delivery past the cutoff (Retry-After may still fire it)', async () => {
+    const methods = await runReconcile(joblessRun(OLD()), async () => ({ status: 'pending' }));
+    expect(methods.recordRunOutcome).not.toHaveBeenCalled();
+  });
+
+  it('does NOT orphan a LEASED delivery past the cutoff', async () => {
+    const methods = await runReconcile(joblessRun(OLD()), async () => ({ status: 'leased' }));
+    expect(methods.recordRunOutcome).not.toHaveBeenCalled();
+  });
+
+  it('does NOT orphan a STAGING delivery past the cutoff', async () => {
+    const methods = await runReconcile(joblessRun(OLD()), async () => ({ status: 'staging' }));
+    expect(methods.recordRunOutcome).not.toHaveBeenCalled();
+  });
+
+  it('falls back to the legacy interrupted orphan for a SUCCEEDED delivery past the cutoff', async () => {
+    const methods = await runReconcile(joblessRun(OLD()), async () => ({ status: 'succeeded' }));
+    expect(methods.recordRunOutcome).toHaveBeenCalledWith(
+      expect.objectContaining({ scheduleId: 's1', status: 'interrupted' }),
+    );
+  });
+
+  it('does not prematurely orphan a SUCCEEDED delivery before the cutoff', async () => {
+    const methods = await runReconcile(joblessRun(YOUNG()), async () => ({ status: 'succeeded' }));
+    expect(methods.recordRunOutcome).not.toHaveBeenCalled();
+  });
+
+  it('uses the legacy orphan policy when the delivery record is gone (null)', async () => {
+    const methods = await runReconcile(joblessRun(OLD()), async () => null);
+    expect(methods.recordRunOutcome).toHaveBeenCalledWith(
+      expect.objectContaining({ status: 'interrupted' }),
+    );
+  });
+
+  it('DEFERS (stays reconcilable) when the delivery lookup fails', async () => {
+    const methods = await runReconcile(joblessRun(OLD()), async () => {
+      throw new Error('delivery store unavailable');
+    });
+    expect(methods.recordRunOutcome).not.toHaveBeenCalled();
+  });
+});
+
 describe('reconciliation is isolated per row', () => {
   /**
    * getRunsForReconciliation returns the OLDEST rows first, so a row that always throws

packages/api/src/schedules/engine.ts

@@ -218,20 +218,60 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine {
               await clearRetainedJob();
               continue;
             }
-            // A `started` run whose job is gone (jobStatus null) is an orphan. Every run
-            // records its conversationId up front, so getJobStatus above already
-            // liveness-checked it: a live long-running fire reads as `running` and is left
-            // alone. v1 is single-process, so a null job genuinely means gone. A fresh
-            // abort also DELETES the job (account-deletion quiesce), so absence carries
-            // the same abort fence as `aborted` above.
+            // A `started` run whose job is gone (jobStatus null). Every run records its
+            // conversationId up front, so getJobStatus above already liveness-checked it: a
+            // live long-running fire reads as `running` and is left alone. A fresh abort
+            // also DELETES the job (account-deletion quiesce), so absence carries the same
+            // abort fence as `aborted` above.
+            //
+            // But a jobless run is NOT automatically an orphan: fireSchedule reserves the
+            // run and its capacity BEFORE the durable trigger delivery reaches the chat
+            // route, and that delivery can be deferred (a `Retry-After` up to 24h, far past
+            // the 30-minute cutoff) or dead-lettered by a pre-generation rejection (an
+            // interactive limiter, PII, or moderation). Consult the durable delivery keyed
+            // by the reservation before deciding.
             if (
               run.status === 'started' &&
               jobStatus == null &&
-              ageMs > ORPHAN_RUN_AGE_MS &&
               !hasAbortInFlight(run, Date.now()) &&
               !hasResumeHandoffInFlight(run, Date.now())
             ) {
-              await finalize('interrupted');
+              let delivery: Awaited<ReturnType<typeof deps.getTriggerDelivery>> | null = null;
+              try {
+                delivery = run.deliveryKey ? await deps.getTriggerDelivery(run.deliveryKey) : null;
+              } catch (deliveryError) {
+                // Unknown is not gone: defer rather than orphan a possibly-live delivery.
+                logger.warn(
+                  `[schedules] delivery lookup failed for ${run.scheduleId}@${run.scheduledFor?.toISOString?.() ?? run.scheduledFor}; deferring:`,
+                  deliveryError,
+                );
+                continue;
+              }
+              const deliveryStatus = delivery?.status;
+              if (
+                deliveryStatus === 'staging' ||
+                deliveryStatus === 'pending' ||
+                deliveryStatus === 'leased'
+              ) {
+                // Admission is still live; the delivery will fire or dead-letter later.
+                continue;
+              }
+              if (deliveryStatus === 'dead') {
+                // Definitively failed before a generation ever existed. Record the error
+                // PROMPTLY from the durable lastError and release capacity through the
+                // ordinary outcome path — no need to wait the 30-minute orphan age.
+                await finalize(
+                  'error',
+                  delivery?.lastError ?? 'Scheduled delivery failed before running',
+                );
+                continue;
+              }
+              // `succeeded` (a generation ran; its own job/outcome evidence is the
+              // authority) or NO delivery record (a legacy fire): only a genuinely aged
+              // jobless run is an orphan.
+              if (ageMs > ORPHAN_RUN_AGE_MS) {
+                await finalize('interrupted');
+              }
               continue;
             }
             if (

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

@@ -2,6 +2,7 @@ import type { ScheduleEngineDeps, ScheduleLimits, ScheduleUserContext } from './
 import type { FireableSchedule } from './types';
 import { AgentTriggerServiceUnavailableError } from '../agents/triggers/service';
 import { AgentTriggerDeliveryError } from '../agents/triggers/delivery';
+import { getAgentTriggerIdempotencyKey } from '../agents/triggers/envelope';
 import { buildFireClientRequestId, fireSchedule } from './fire';
 import { withCapacitySlot } from './capacity';
 
@@ -86,6 +87,7 @@ function makeMethods() {
         scheduledFor: Date;
         conversationId?: string;
         capacitySlot?: number;
+        deliveryKey?: string;
       }) => {
         const k = key(data.scheduleId, data.scheduledFor);
         if (runs.has(k)) {
@@ -284,6 +286,30 @@ describe('fireSchedule', () => {
     expect(methods.releaseLeaseByHolder).toHaveBeenCalledWith('sched-1', 'claim-holder-1');
   });
 
+  it('stamps the deterministic deliveryKey on the reservation (== the enqueued key)', async () => {
+    const { methods } = makeMethods();
+    const enqueueTrigger = jest.fn<
+      ReturnType<ScheduleEngineDeps['enqueueTrigger']>,
+      Parameters<ScheduleEngineDeps['enqueueTrigger']>
+    >(async () => undefined);
+
+    const result = await fireSchedule(
+      makeDeps(methods, { enqueueTrigger }),
+      makeSchedule(),
+      LIMITS,
+      dueAt(),
+    );
+
+    expect(result.fired).toBe(true);
+    const reservation = (methods.reserveStartedRun as jest.Mock).mock.calls[0][0];
+    const envelope = enqueueTrigger.mock.calls[0][0];
+    // Stored BEFORE enqueue and equal to the enqueued delivery's idempotency key, so
+    // reconciliation can read the durable delivery for THIS occurrence even if the enqueue
+    // committed ambiguously.
+    expect(reservation.deliveryKey).toEqual(expect.any(String));
+    expect(reservation.deliveryKey).toBe(getAgentTriggerIdempotencyKey(envelope));
+  });
+
   it('marks a Run Now trigger as manual for downstream limiter policy', async () => {
     const { methods } = makeMethods();
     const enqueueTrigger = jest.fn<

packages/api/src/schedules/fire.ts

@@ -2,7 +2,12 @@ import { randomUUID } from 'node:crypto';
 import { logger } from '@librechat/data-schemas';
 import type { ScheduleEngineDeps, ScheduleLimits, FireResult, FireableSchedule } from './types';
 import type { JsonValue } from '../agents/json';
-import { AgentTriggerEnvelopeError, createAgentTriggerEnvelope } from '../agents/triggers/envelope';
+import type { AgentTriggerEnvelope } from '../agents/triggers/envelope';
+import {
+  AgentTriggerEnvelopeError,
+  createAgentTriggerEnvelope,
+  getAgentTriggerIdempotencyKey,
+} from '../agents/triggers/envelope';
 import { AgentTriggerServiceUnavailableError } from '../agents/triggers/service';
 import { AgentTriggerDeliveryError } from '../agents/triggers/delivery';
 import { computeNextRunAt, cadenceIntervalMinutes } from './cadence';
@@ -32,15 +37,21 @@ class ScheduleFireError extends Error {
   }
 }
 
-async function enqueueScheduleTrigger(
-  deps: ScheduleEngineDeps,
+/**
+ * Builds the durable trigger envelope for one occurrence. Pure and deterministic in the
+ * fields that {@link getAgentTriggerIdempotencyKey} hashes (only `requestId`/`receivedAt`
+ * vary), so the `deliveryKey` derived from it is stable across retries and is stored on
+ * the reservation BEFORE enqueue — that lets schedule reconciliation read the durable
+ * delivery state even for an ambiguously-committed enqueue.
+ */
+function buildScheduleTriggerEnvelope(
   schedule: FireableSchedule,
   user: NonNullable<Awaited<ReturnType<ScheduleEngineDeps['getUserContext']>>>,
   scheduledFor: Date,
   files: Awaited<ReturnType<ScheduleEngineDeps['resolveFiles']>>,
   conversationId: string,
   manual: boolean,
-): Promise<void> {
+): AgentTriggerEnvelope {
   const occurrenceId = buildFireClientRequestId(schedule.id, scheduledFor);
   const triggerFiles: JsonValue[] = files.map((file) => {
     const value: { [key: string]: JsonValue } = { file_id: file.file_id };
@@ -52,7 +63,7 @@ async function enqueueScheduleTrigger(
     if (file.source != null) value.source = file.source;
     return value;
   });
-  const envelope = createAgentTriggerEnvelope({
+  return createAgentTriggerEnvelope({
     mode: 'fire',
     requestId: randomUUID(),
     deliveryId: occurrenceId,
@@ -82,8 +93,15 @@ async function enqueueScheduleTrigger(
       },
     },
   });
+}
+
+async function enqueueScheduleTrigger(
+  deps: ScheduleEngineDeps,
+  envelope: AgentTriggerEnvelope,
+  orderingKey: string,
+): Promise<void> {
   try {
-    await deps.enqueueTrigger(envelope, { orderingKey: schedule.id });
+    await deps.enqueueTrigger(envelope, { orderingKey });
   } catch (error) {
     const message = error instanceof Error ? error.message : String(error);
     const definite =
@@ -374,6 +392,35 @@ export async function fireSchedule(
     // rejects a second `started` run for the schedule, so a concurrent occurrence
     // surfaces as 'overlap' with no read-then-insert race.
     const conversationId = randomUUID();
+    // Build the durable trigger envelope NOW so the deterministic deliveryKey can be
+    // stored on the reservation below — before enqueue. That way schedule reconciliation
+    // can read the delivery state (live vs dead) even when the enqueue commits ambiguously,
+    // instead of orphaning a still-live delivery after the 30-minute cutoff. A schedule
+    // whose envelope cannot be built can never fire, so skip WITHOUT reserving a phantom
+    // `started` run (a definite, retry-invariant failure).
+    let triggerEnvelope: AgentTriggerEnvelope;
+    try {
+      triggerEnvelope = buildScheduleTriggerEnvelope(
+        schedule,
+        user,
+        scheduledFor,
+        files,
+        conversationId,
+        options?.manual === true,
+      );
+    } catch (envelopeError) {
+      logger.error(`[schedules] trigger envelope build failed for ${schedule.id}:`, envelopeError);
+      if (options?.manual) {
+        await releaseManualLease();
+      } else {
+        await advance();
+      }
+      return {
+        fired: false,
+        error: envelopeError instanceof Error ? envelopeError.message : String(envelopeError),
+      };
+    }
+    const deliveryKey = getAgentTriggerIdempotencyKey(triggerEnvelope);
     // The GLOBAL fireConcurrency cap is enforced by claiming a unique capacity slot in
     // the SAME insert that reserves the run, so it is decided by the DB rather than by
     // a count read before the write. The allocator advances to the next free slot when
@@ -395,6 +442,7 @@ export async function fireSchedule(
           conversationId,
           firedAt: new Date(),
           capacitySlot,
+          deliveryKey,
           ...(typeof schedule.configRevision === 'number'
             ? { configRevision: schedule.configRevision }
             : {}),
@@ -500,15 +548,7 @@ export async function fireSchedule(
     }
 
     try {
-      await enqueueScheduleTrigger(
-        deps,
-        schedule,
-        user,
-        scheduledFor,
-        files,
-        conversationId,
-        options?.manual === true,
-      );
+      await enqueueScheduleTrigger(deps, triggerEnvelope, schedule.id);
     } catch (error) {
       const message = error instanceof Error ? error.message : String(error);
       const ambiguous = error instanceof ScheduleFireError && error.ambiguous;

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

@@ -34,6 +34,7 @@ function makeService(base: Cfg, merged: Cfg = base) {
     resolveAgentFireAccess: jest.fn(async () => 'ok' as const),
     isUserDeleting: jest.fn(async () => false),
     enqueueAgentTrigger: jest.fn(async () => undefined),
+    getTriggerDelivery: jest.fn(async () => null),
   } as unknown as SchedulesServiceDeps;
   return createSchedulesService(deps);
 }

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

@@ -65,6 +65,7 @@ function makeService(
     resolveAgentFireAccess: jest.fn(async () => 'ok' as const),
     isUserDeleting: jest.fn(async () => false),
     enqueueAgentTrigger,
+    getTriggerDelivery: jest.fn(async () => null),
   } as unknown as SchedulesServiceDeps;
   // Short bounded waits so fail-closed drain paths resolve in test time.
   return createSchedulesService(deps, {
@@ -166,6 +167,7 @@ describe('balance initialization', () => {
       resolveAgentFireAccess: jest.fn(async () => 'ok' as const),
       isUserDeleting: jest.fn(async () => false),
       enqueueAgentTrigger: jest.fn(async () => undefined),
+      getTriggerDelivery: jest.fn(async () => null),
     } as unknown as SchedulesServiceDeps);
     return { service, upsertBalance, initializeNullBalance, findBalance };
   }
@@ -276,6 +278,7 @@ describe('balance initialization', () => {
       resolveAgentFireAccess: jest.fn(async () => 'ok' as const),
       isUserDeleting: jest.fn(async () => false),
       enqueueAgentTrigger: jest.fn(async () => undefined),
+      getTriggerDelivery: jest.fn(async () => null),
     } as unknown as SchedulesServiceDeps);
 
     await service.engineDeps.isOutOfBalance({ id: 'user-1' } as never);
@@ -1210,6 +1213,7 @@ describe('scheduled resume capacity', () => {
       resolveAgentFireAccess: jest.fn(async () => 'ok' as const),
       isUserDeleting: jest.fn(async () => false),
       enqueueAgentTrigger: jest.fn(async () => undefined),
+      getTriggerDelivery: jest.fn(async () => null),
     } as unknown as SchedulesServiceDeps);
     return { service, methods };
   }

packages/api/src/schedules/service.ts

@@ -164,6 +164,9 @@ export interface SchedulesServiceDeps {
   isUserDeleting: (userId: string) => Promise<boolean>;
   /** Shared durable trigger admission from the merged agent-trigger service. */
   enqueueAgentTrigger: ScheduleEngineDeps['enqueueTrigger'];
+  /** Reads a durable trigger delivery by key from the merged agent-trigger service, for
+   *  reconciliation's live-vs-dead delivery check. */
+  getTriggerDelivery: ScheduleEngineDeps['getTriggerDelivery'];
 }
 
 export interface SchedulesService {
@@ -292,6 +295,7 @@ export function createSchedulesService(
     'resolveAgentFireAccess',
     'isUserDeleting',
     'enqueueAgentTrigger',
+    'getTriggerDelivery',
   ];
   for (const key of REQUIRED_DEPS) {
     if (deps[key] == null) {
@@ -501,6 +505,7 @@ export function createSchedulesService(
       }));
     },
     enqueueTrigger: deps.enqueueAgentTrigger,
+    getTriggerDelivery: deps.getTriggerDelivery,
     runInTenantContext: (user, fn) =>
       tenantStorage.run({ tenantId: user.tenantId, userId: user.id }, fn),
     getJobStatus: async (conversationId) => {

packages/api/src/schedules/types.ts

@@ -1,4 +1,8 @@
-import type { ScheduleMethods, ISchedule } from '@librechat/data-schemas';
+import type {
+  ScheduleMethods,
+  ISchedule,
+  AgentTriggerDeliveryStatus,
+} from '@librechat/data-schemas';
 import type { Types } from 'mongoose';
 import type { AgentTriggerEnqueueOptions, AgentTriggerEnvelope } from '../agents/triggers';
 import type { SlotClaimResult } from './capacity';
@@ -137,6 +141,15 @@ export interface ScheduleEngineDeps {
     envelope: AgentTriggerEnvelope,
     options?: AgentTriggerEnqueueOptions,
   ) => Promise<unknown>;
+  /**
+   * Reads the durable trigger delivery for a reservation's `deliveryKey`, so
+   * reconciliation can tell a still-live admission (`staging`/`pending`/`leased`, which a
+   * `Retry-After` can defer up to 24h) or a dead-letter (`dead`, with its `lastError`)
+   * apart from a genuinely orphaned jobless run. Null when no delivery record exists.
+   */
+  getTriggerDelivery: (
+    deliveryKey: string,
+  ) => Promise<{ status: AgentTriggerDeliveryStatus; lastError?: string } | null>;
   /** Runs fn inside the owner's tenant ALS context. */
   runInTenantContext: <T>(user: ScheduleUserContext, fn: () => Promise<T>) => Promise<T>;
   /**

packages/data-schemas/src/schema/scheduleRun.ts

@@ -29,6 +29,12 @@ const scheduleRunSchema: Schema<IScheduleRunDocument> = new Schema(
     conversationId: {
       type: String,
     },
+    /** Deterministic durable-trigger delivery key for this occurrence, stamped at
+     *  reservation before enqueue so reconciliation can read the delivery's live/dead
+     *  state rather than orphan-settling a deferred or dead-lettered run. */
+    deliveryKey: {
+      type: String,
+    },
     /** Fresh while a RESUME of this paused run is mid-flight; a re-pause hand-off's
      *  writes are still landing, so quiesce must not settle on the paused job state.
      *  Cleared by the pause record (the hand-off completion) or aged out. */

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

@@ -83,6 +83,11 @@ export interface IScheduleRun {
   conversationId?: string;
   status: ScheduleRunStatus;
   error?: string;
+  /** Deterministic durable-trigger delivery key for this occurrence, stamped at
+   *  reservation (before enqueue). Lets reconciliation read the delivery's live/dead
+   *  state instead of orphan-settling a jobless run that is merely deferred (Retry-After)
+   *  or that dead-lettered before a generation ever started. */
+  deliveryKey?: string;
   droppedFileIds?: string[];
   durationMs?: number;
   bookkept?: boolean;