🎡 fix: Scheduled Chat Slot Accounting and Reconciliation Rotation (#15040)

* fix(schedules): harden limits and reconciliation

* docs(schedules): align reconciliation invariants

* fix(schedules): preserve prompt editor contract

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

@@ -314,9 +314,9 @@ describe('reconciliation consults the durable trigger delivery', () => {
 
 describe('reconciliation is isolated per row', () => {
   /**
-   * getRunsForReconciliation returns the OLDEST rows first, so a row that always throws
-   * came back every tick and starved every run behind it — in the one component whose
-   * entire job is settling states nothing else will.
+   * A row that always throws must not abort the pass. The store stamps every examined
+   * row afterward so persistent failures rotate behind rows the bounded window has not
+   * inspected yet.
    */
   it('keeps reconciling after a row throws', async () => {
     const methods = makeMethods(makeClaimedSchedule());

packages/api/src/schedules/engine.ts

@@ -63,11 +63,10 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine {
       );
       await runAsSystem(async () => {
         for (const run of runs) {
-          // PER-ROW isolation. A single throwing row used to abort the whole pass, and
-          // since this query returns the OLDEST rows first, that row came back every
-          // tick and starved every run behind it indefinitely. Reconciliation is the
-          // backstop for exactly the states nothing else settles, so it has to make
-          // progress on the rest.
+          // PER-ROW isolation. A single throwing row used to abort the whole pass.
+          // Reconciliation is the backstop for exactly the states nothing else
+          // settles, so it has to make progress on the rest; the examined-at stamp
+          // below then rotates failures behind rows this pass did not inspect.
           try {
             // Identity-fence the job lookup: a replacement user turn reuses this
             // conversationId but sheds the scheduleId/scheduledFor metadata. Only

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

@@ -376,6 +376,20 @@ describe('createScheduleWithSlot (atomic per-user cap)', () => {
     expect(c).not.toBe('limit');
     expect(await methods.countSchedulesByUser(user)).toBe(2);
   });
+
+  it('counts legacy schedules without slots against the cap', async () => {
+    const user = new mongoose.Types.ObjectId();
+    await methods.createSchedule(scheduleData({ user }));
+    await methods.createSchedule(scheduleData({ user }));
+
+    const results = await Promise.all(
+      Array.from({ length: 3 }, () => methods.createScheduleWithSlot(scheduleData({ user }), 3)),
+    );
+
+    expect(results.filter((result) => result !== 'limit')).toHaveLength(1);
+    expect(results.filter((result) => result === 'limit')).toHaveLength(2);
+    expect(await methods.countSchedulesByUser(user)).toBe(3);
+  });
 });
 
 describe('recordRunOutcome', () => {
@@ -2423,6 +2437,31 @@ describe('reconciliation rotates the paused window', () => {
   });
 });
 
+describe('reconciliation rotates the started window', () => {
+  it('serves a started row behind a full window once the leaders have been examined', async () => {
+    const user = new mongoose.Types.ObjectId();
+    const base = Date.parse('2026-07-01T00:00:00Z');
+    const olderThan = new Date(base + 10_000_000);
+    const runs = Array.from({ length: 101 }, (_, index) => ({
+      scheduleId: `started_${index}`,
+      user,
+      scheduledFor: new Date(base + index * 60_000),
+      firedAt: new Date(base + index * 60_000),
+      status: 'started' as const,
+    }));
+    await ScheduleRun.insertMany(runs);
+
+    const first = await methods.getRunsForReconciliation(olderThan, 100);
+    const queuedAt = runs[100].scheduledFor;
+    expect(first.some((run) => run.scheduledFor?.getTime() === queuedAt.getTime())).toBe(false);
+
+    await methods.markRunsReconciled(first);
+
+    const second = await methods.getRunsForReconciliation(olderThan, 100);
+    expect(second.some((run) => run.scheduledFor?.getTime() === queuedAt.getTime())).toBe(true);
+  });
+});
+
 describe('requestRunAbort (renewable, source-aware stamp)', () => {
   const scheduledFor = new Date('2026-07-20T12:00:00Z');
 

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

@@ -339,7 +339,9 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
       const taken = new Set(
         used.map((s) => s.slot).filter((s): s is number => typeof s === 'number'),
       );
-      if (taken.size >= maxPerUser) {
+      // Every live row consumes capacity, including legacy/internal rows created
+      // before the slot allocator existed. Slots remain the atomic collision key.
+      if (used.length >= maxPerUser) {
         return 'limit';
       }
       let slot = 0;
@@ -1505,16 +1507,17 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
   /**
    * Non-terminal runs old enough to need a job-store status check. Fetches
    * `started` (capacity-consuming) and `requires_action` (paused) in separate
-   * budgeted, firedAt-ordered buckets so a backlog of long-lived paused rows
-   * can't starve orphaned `started` runs out of every sweep.
+   * budgeted round-robin buckets so a backlog of live rows in either state
+   * cannot starve an orphaned run out of every sweep.
    */
   async function getRunsForReconciliation(olderThan: Date, limit: number): Promise<IScheduleRun[]> {
     const [started, paused] = await Promise.all([
-      // `started` runs are bounded by the global fireConcurrency cap, so this window
-      // can never fill with rows that have nothing to do — oldest-first is right here.
+      // A deployment may intentionally set fireConcurrency above the reconciliation
+      // batch. Rotate started rows as well, otherwise a full oldest-first window of
+      // legitimate long-running generations can hide a newer abandoned run forever.
       ScheduleRun()
         .find({ status: 'started', firedAt: { $lt: olderThan } })
-        .sort({ firedAt: 1 })
+        .sort({ reconciledAt: 1, firedAt: 1 })
         .limit(limit)
         .lean<IScheduleRun[]>(),
       // ROUND-ROBIN, not oldest-first. A paused run holds no capacity slot and does not
@@ -1534,8 +1537,8 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
     return [...started, ...paused];
   }
 
-  /** Stamps rows as examined, so the paused window rotates instead of re-serving the
-   *  same rows forever. Bookkeeping only: never touches `updatedAt`. */
+  /** Stamps rows as examined, so each bounded reconciliation window rotates instead
+   *  of re-serving the same rows forever. Bookkeeping only: never touches `updatedAt`. */
   async function markRunsReconciled(runs: Array<Pick<IScheduleRun, '_id'>>): Promise<void> {
     const ids = runs.map((run) => run._id).filter((id) => id != null);
     if (ids.length === 0) {

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

@@ -96,8 +96,8 @@ const scheduleRunSchema: Schema<IScheduleRunDocument> = new Schema(
     abortPersistedAt: {
       type: Date,
     },
-    /** When reconciliation last examined this row. Orders the paused window so a full
-     *  batch of still-live pauses cannot starve an abandoned row behind them. */
+    /** When reconciliation last examined this row. Rotates each bounded non-terminal
+     *  window so a full batch of live runs cannot starve an abandoned row behind it. */
     reconciledAt: {
       type: Date,
     },
@@ -147,8 +147,8 @@ scheduleRunSchema.index({ scheduleId: 1, firedAt: -1 });
 // Reconciliation sweeps by status; keeps `started` (capacity) fetch cheap and
 // prevents long-lived `requires_action` rows from starving the scan.
 scheduleRunSchema.index({ status: 1, firedAt: 1 });
-// The paused reconciliation window sorts on {reconciledAt, firedAt} within a status;
-// without this the round-robin rotation re-sorts the whole paused set every tick.
+// Non-terminal reconciliation windows sort on {reconciledAt, firedAt} within a status;
+// without this the round-robin rotation re-sorts the whole live set every tick.
 scheduleRunSchema.index({ status: 1, reconciledAt: 1, firedAt: 1 });
 
 export default scheduleRunSchema;

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

@@ -107,8 +107,8 @@ export interface IScheduleRun {
   abortPersistedAt?: Date;
   /** The schedule's configRevision at claim time. */
   configRevision?: number;
-  /** When reconciliation last examined this row; orders the paused window so no row
-   *  can be starved by a full batch of still-live pauses ahead of it. */
+  /** When reconciliation last examined this row; rotates each bounded non-terminal
+   *  window so no abandoned row can starve behind a full batch of live runs. */
   reconciledAt?: Date;
   resumeClaimedAt?: Date;
   createdAt?: Date;