From 69d863fbee69f6358940e84d86c55bdef3e508b0 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 18:04:18 +0000 Subject: [PATCH] =?UTF-8?q?fix(schedules):=20Wave=20A=20=E2=80=94=20null-b?= =?UTF-8?q?alance=20CAS,=20atomic=20paused-card=20clear,=20clustered=20era?= =?UTF-8?q?sure=20sweep?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Slice 1 (thread r3804518381): route the existing-null balance initialization through a { user, tokenCredits: null } compare-and-set instead of a blind $set, so a concurrent initializer/charge landing between the preflight read and the write is never handed back its spent starting balance. On a CAS miss the preflight re-reads the winner. The absent-record $setOnInsert path and the credited-record refill-config sync are unchanged. Adds the initializeNullBalance adapter (no upsert) and regression coverage for winner/miss/sync cases. Slice 2 (thread r3804518388): updateScheduleById now drops a `requires_action` lastRun projection atomically with the configRevision bump. Any pause present at edit time was projected under the pre-edit revision and can never be replaced by its own revision-fenced terminal outcome — a disabling edit would strand the card on "Needs approval" forever. Implemented as classic-operator CAS branches (DocumentDB rules out a conditional pipeline $unset), fenced on the card STILL being the pause so a terminal outcome or newer occurrence that races in is preserved. Terminal history survives untouched. Slice 4 (thread r3803826204): expose initializeScheduleErasureSweep from the schedule runtime facade and start it in every clustered (experimental) worker after Mongo is up. It re-drives eraseScheduleIfDrained for soft-deleted rows so a hidden prompt cannot outlive its drain when the delete/erase-on-settle attempts miss. It arms nothing else and never infers owner death from a process-local missing job (isTopologySafeToArm gates that). Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq --- api/server/experimental.js | 12 +- api/server/experimental.spec.js | 18 +++ api/server/services/Schedules/index.js | 10 ++ packages/api/src/schedules/gate.spec.ts | 2 + packages/api/src/schedules/service.spec.ts | 109 ++++++++++++++++- packages/api/src/schedules/service.ts | 82 ++++++++++--- .../src/methods/schedule.methods.spec.ts | 115 ++++++++++++++++++ packages/data-schemas/src/methods/schedule.ts | 60 ++++++--- 8 files changed, 362 insertions(+), 46 deletions(-) diff --git a/api/server/experimental.js b/api/server/experimental.js index 2c4cfe97c3..13cddf9de4 100644 --- a/api/server/experimental.js +++ b/api/server/experimental.js @@ -37,7 +37,10 @@ const createValidateImageRequest = require('./middleware/validateImageRequest'); const { startExpiredFileSweep } = require('./services/Files/process'); const { initializeGitHubSkillSync } = require('./services/Skills/sync'); const { initializeAgentTriggerService } = require('./services/Agents/triggers'); -const { recordExpiredScheduleApproval } = require('./services/Schedules'); +const { + recordExpiredScheduleApproval, + initializeScheduleErasureSweep, +} = require('./services/Schedules'); const { jwtLogin, ldapLogin, passportLogin } = require('~/strategies'); const { updateInterfacePermissions: updateInterfacePerms } = require('@librechat/api'); const { @@ -335,6 +338,13 @@ if (cluster.isMaster) { logger.error(`[Worker ${process.pid}][indexSync] Background sync failed:`, err); }); + // This entrypoint deliberately does not arm the schedule engine, but DELETE stays + // open — so soft-deleted rows still accrue with no reconciler to erase them. Start + // the erasure-ONLY sweep (Mongo is up, GenerationJobManager was initialized above): + // it never claims, fires, advances, or infers owner death from a process-local + // missing job, and its idempotent guard makes this safe once per worker. + initializeScheduleErasureSweep(); + app.disable('x-powered-by'); app.set('trust proxy', trusted_proxy); diff --git a/api/server/experimental.spec.js b/api/server/experimental.spec.js index bc6047c921..6a90fa0bd8 100644 --- a/api/server/experimental.spec.js +++ b/api/server/experimental.spec.js @@ -33,6 +33,24 @@ describe('Experimental server configuration', () => { expect(initializeIndex).toBeGreaterThan(handlerIndex); }); + it('starts erasure-only schedule maintenance after connecting to Mongo, once per worker', () => { + const connectIndex = source.indexOf('await connectDb();'); + const sweepIndex = source.indexOf('initializeScheduleErasureSweep();'); + + expect(connectIndex).toBeGreaterThan(-1); + expect(sweepIndex).toBeGreaterThan(-1); + // Mongo must be up before the sweep reads soft-deleted rows. + expect(sweepIndex).toBeGreaterThan(connectIndex); + // Idempotent guard lives in the service; started exactly once from this entrypoint. + expect(source.match(/initializeScheduleErasureSweep\(\);/g)).toHaveLength(1); + }); + + it('never arms the full schedule engine in a clustered worker', () => { + // The clustered entrypoint runs erasure-only maintenance: arming the engine here + // would claim/fire/absence-reconcile runs whose peer generations it cannot see. + expect(source).not.toContain('initializeScheduleEngine('); + }); + it('runs cross-tenant startup work in the system context', () => { expect(source).toContain('await runAsSystem(seedDatabase);'); expect(source).toMatch( diff --git a/api/server/services/Schedules/index.js b/api/server/services/Schedules/index.js index ba051f34fb..6f425ebf85 100644 --- a/api/server/services/Schedules/index.js +++ b/api/server/services/Schedules/index.js @@ -35,6 +35,15 @@ function getService() { }, { upsert: true, new: true }, ).lean(), + // Compare-and-set: only initialize an existing record while its credit is still null. + // No upsert — a CAS miss must re-read the winner, not insert a fresh balance. `null` + // in the filter also matches a legacy record whose `tokenCredits` field is absent. + initializeNullBalance: (userId, { tokenCredits, sync }) => + mongoose.models.Balance.findOneAndUpdate( + { user: userId, tokenCredits: null }, + { $set: { tokenCredits, ...(sync && Object.keys(sync).length > 0 ? sync : {}) } }, + { new: true }, + ).lean(), enqueueAgentTrigger, resolveAgentFireAccess, // Reuse the merged trigger/deletion admission fence. A schedule fire and every @@ -82,6 +91,7 @@ module.exports = { deleteScheduleForOwner: invoke('deleteScheduleForOwner'), quiesceUserSchedules: invoke('quiesceUserSchedules'), initializeScheduleEngine: invoke('initializeScheduleEngine'), + initializeScheduleErasureSweep: invoke('initializeScheduleErasureSweep'), recordExpiredScheduleApproval, isUserDeleting: async (userId) => { const methods = require('~/models'); diff --git a/packages/api/src/schedules/gate.spec.ts b/packages/api/src/schedules/gate.spec.ts index 015208796f..c4ab51301a 100644 --- a/packages/api/src/schedules/gate.spec.ts +++ b/packages/api/src/schedules/gate.spec.ts @@ -30,6 +30,7 @@ function makeService(base: Cfg, merged: Cfg = base) { findUserById: jest.fn(async () => null), findBalance: jest.fn(async () => null), upsertBalance: jest.fn(async () => null), + initializeNullBalance: jest.fn(async () => null), resolveAgentFireAccess: jest.fn(async () => 'ok' as const), isUserDeleting: jest.fn(async () => false), enqueueAgentTrigger: jest.fn(async () => undefined), @@ -152,6 +153,7 @@ describe('v1 experimental gate, asserted at real entry points', () => { findUserById: jest.fn(), findBalance: jest.fn(), upsertBalance: jest.fn(), + initializeNullBalance: jest.fn(), resolveAgentFireAccess: jest.fn(), } as unknown as SchedulesServiceDeps), ).toThrow(/isUserDeleting/); diff --git a/packages/api/src/schedules/service.spec.ts b/packages/api/src/schedules/service.spec.ts index 2d65ac7e63..ee4730f673 100644 --- a/packages/api/src/schedules/service.spec.ts +++ b/packages/api/src/schedules/service.spec.ts @@ -58,6 +58,7 @@ function makeService( findUserById: jest.fn(async () => null), findBalance: jest.fn(async () => null), upsertBalance: jest.fn(async () => null), + initializeNullBalance: jest.fn(async () => null), resolveAgentFireAccess: jest.fn(async () => 'ok' as const), isUserDeleting: jest.fn(async () => false), enqueueAgentTrigger, @@ -139,19 +140,29 @@ describe('balance initialization', () => { balance: { enabled: true, startBalance: 20000 }, } as unknown as Awaited>; - function serviceWithBalance(existing: Record | null) { + function serviceWithBalance( + existing: Record | null, + overrides: { + initializeNullBalance?: jest.Mock; + findBalance?: jest.Mock; + } = {}, + ) { const upsertBalance = jest.fn(async () => ({ tokenCredits: 20000 })); + const initializeNullBalance = + overrides.initializeNullBalance ?? jest.fn(async () => ({ tokenCredits: 20000 })); + const findBalance = overrides.findBalance ?? jest.fn(async () => existing); const service = createSchedulesService({ methods: {} as unknown as SchedulesServiceDeps['methods'], getAppConfig: (async () => balanceConfig) as SchedulesServiceDeps['getAppConfig'], findUserById: jest.fn(async () => null), - findBalance: jest.fn(async () => existing), + findBalance, upsertBalance, + initializeNullBalance, resolveAgentFireAccess: jest.fn(async () => 'ok' as const), isUserDeleting: jest.fn(async () => false), enqueueAgentTrigger: jest.fn(async () => undefined), } as unknown as SchedulesServiceDeps); - return { service, upsertBalance }; + return { service, upsertBalance, initializeNullBalance, findBalance }; } const updateFrom = (spy: jest.Mock) => @@ -162,6 +173,14 @@ describe('balance initialization', () => { ] )[1]; + const casUpdateFrom = (spy: jest.Mock) => + ( + spy.mock.calls[0] as unknown as [ + string, + { tokenCredits: number; sync: Record }, + ] + )[1]; + /** * The balance READ and this write are separate statements. A concurrent charge that * creates the record in between would be overwritten by a blind `$set`, handing back @@ -178,13 +197,90 @@ describe('balance initialization', () => { expect(update.set).not.toHaveProperty('tokenCredits'); }); - it('still $sets a null credit on an EXISTING record, which has no charge to clobber', async () => { - const { service, upsertBalance } = serviceWithBalance({ autoRefillEnabled: false }); + /** + * The record was observed with a null credit, but the read and the write are separate + * statements: a concurrent initializer/charge could have set and spent it in between. + * The credit must go through a `{ tokenCredits: null }` CAS, never a blind `$set` that + * would restore spent credits. + */ + it('initializes an EXISTING null credit through the CAS, never a blind $set', async () => { + const { service, upsertBalance, initializeNullBalance } = serviceWithBalance({ + autoRefillEnabled: false, + }); await service.engineDeps.isOutOfBalance({ id: 'user-1' } as never); + expect(initializeNullBalance).toHaveBeenCalledTimes(1); + const cas = casUpdateFrom(initializeNullBalance); + expect(cas.tokenCredits).toBe(20000); + // The credit write goes ONLY through the CAS, never the unconditional upsert path. + expect(upsertBalance).not.toHaveBeenCalled(); + }); + + /** + * When the CAS misses — a concurrent initializer/charge won — the preflight must re-read + * the winner's balance rather than restore the starting credit. + */ + it('re-reads the winner on a CAS miss and never restores spent credits', async () => { + const initializeNullBalance = jest.fn(async () => null); + // First read observes a null credit; the winner's re-read returns the spent balance. + const findBalance = jest + .fn() + .mockResolvedValueOnce({ autoRefillEnabled: false, tokenCredits: null }) + .mockResolvedValueOnce({ tokenCredits: 5 }); + const { service } = serviceWithBalance(null, { initializeNullBalance, findBalance }); + + const outOfBalance = await service.engineDeps.isOutOfBalance({ id: 'user-1' } as never); + + expect(initializeNullBalance).toHaveBeenCalledTimes(1); + // Re-read after the miss (the two isOutOfBalance reads), and the small winner balance + // wins: > 0 credits means the user is not pre-skipped for balance. + expect(findBalance).toHaveBeenCalledTimes(2); + expect(outOfBalance).toBe(false); + }); + + /** + * A stale record whose credit is already set but whose refill config drifted still syncs + * that config, and the sync must not widen into a credit write. + */ + it('syncs refill config on a credited record without writing tokenCredits', async () => { + const refillConfig = { + interfaceConfig: {}, + balance: { + enabled: true, + startBalance: 20000, + autoRefillEnabled: true, + refillIntervalValue: 30, + refillIntervalUnit: 'days', + refillAmount: 10000, + }, + } as unknown as Awaited>; + const upsertBalance = jest.fn(async () => ({ tokenCredits: 100 })); + const initializeNullBalance = jest.fn(async () => null); + const service = createSchedulesService({ + methods: {} as unknown as SchedulesServiceDeps['methods'], + getAppConfig: (async () => refillConfig) as SchedulesServiceDeps['getAppConfig'], + findUserById: jest.fn(async () => null), + findBalance: jest.fn(async () => ({ + tokenCredits: 100, + autoRefillEnabled: false, + lastRefill: new Date('2026-01-01T00:00:00.000Z'), + })), + upsertBalance, + initializeNullBalance, + resolveAgentFireAccess: jest.fn(async () => 'ok' as const), + isUserDeleting: jest.fn(async () => false), + enqueueAgentTrigger: jest.fn(async () => undefined), + } as unknown as SchedulesServiceDeps); + + await service.engineDeps.isOutOfBalance({ id: 'user-1' } as never); + + // Credit is already set, so the CAS path is never taken. + expect(initializeNullBalance).not.toHaveBeenCalled(); + expect(upsertBalance).toHaveBeenCalledTimes(1); const update = updateFrom(upsertBalance); - expect(update.set).toMatchObject({ tokenCredits: 20000 }); + expect(update.set).toMatchObject({ autoRefillEnabled: true }); + expect(update.set).not.toHaveProperty('tokenCredits'); expect(update.setOnInsert).toEqual({}); }); }); @@ -987,6 +1083,7 @@ describe('scheduled resume capacity', () => { findUserById: jest.fn(async () => ({ _id: 'user-1', role: 'USER' })), findBalance: jest.fn(async () => null), upsertBalance: jest.fn(async () => null), + initializeNullBalance: jest.fn(async () => null), resolveAgentFireAccess: jest.fn(async () => 'ok' as const), isUserDeleting: jest.fn(async () => false), enqueueAgentTrigger: jest.fn(async () => undefined), diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index a0c3a5458d..4211dc6668 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -144,6 +144,18 @@ export interface SchedulesServiceDeps { userId: string, update: { set: Partial; setOnInsert: Partial }, ) => Promise; + /** + * Compare-and-set initialization for an EXISTING balance whose `tokenCredits` is still + * null. Writes the starting credit (and any refill-config sync) only while + * `{ user, tokenCredits: null }` still matches — never an upsert — so a concurrent + * initializer/charge that already set credits between the read and this write is not + * clobbered. Returns the winning document, or `null` when the CAS did not match (the + * caller re-reads the winner rather than restoring credits). + */ + initializeNullBalance: ( + userId: string, + update: { tokenCredits: number; sync: Partial }, + ) => Promise; resolveAgentFireAccess: ( agentId: string, user: ScheduleUserContext, @@ -211,6 +223,12 @@ export interface SchedulesService { * the topology cannot be shown safe (process-local job store with no single-process * assertion) — see isTopologySafeToArm. */ initializeScheduleEngine: () => Promise | undefined>; + /** Starts erasure-ONLY maintenance for an entrypoint that never arms the engine (the + * clustered worker). Idempotent per process and a no-op once the full engine is armed. + * Arms nothing else — no claims, firing, cadence advancement, or absence-based + * reconciliation — and refuses to infer owner death from a process-local missing job + * (isTopologySafeToArm gates that). See startScheduleErasureSweep. */ + initializeScheduleErasureSweep: () => void; } /** Test-only overrides for the service's bounded waits (drains, barriers). */ @@ -240,6 +258,7 @@ export function createSchedulesService( 'findUserById', 'findBalance', 'upsertBalance', + 'initializeNullBalance', 'resolveAgentFireAccess', 'isUserDeleting', 'enqueueAgentTrigger', @@ -368,26 +387,38 @@ export function createSchedulesService( if (balanceConfig.startBalance != null) { const updateFields = buildBalanceUpdateFields(balanceConfig, record, user.id); if (Object.keys(updateFields).length > 0) { - // The read above and this write are separate statements, and the credit - // fields are INITIALIZATION values: a concurrent charge that created the - // record in between would be overwritten by a blind `$set`, restoring credits - // the user had already spent. Route those through `$setOnInsert` so they only - // apply to a document this call actually creates. The refill-config fields are - // a genuine sync and stay on `$set`. - // - // Only when the record was ABSENT: an existing record with a null - // tokenCredits has no charge to clobber, so initializing it via `$set` is - // both safe and necessary (`$setOnInsert` would never fire for it). + // The read above and every write below are separate statements, and the credit + // field is an INITIALIZATION value: a concurrent charge that set-and-spent the + // record in between must never be handed back its starting balance. Each case + // fences the credit write so it can only create, never restore, credits. The + // refill-config fields are a genuine sync and never carry a credit. const { user: initUser, tokenCredits, ...syncFields } = updateFields; - const setOnInsert = - record == null - ? { - ...(initUser != null ? { user: initUser } : {}), - ...(tokenCredits != null ? { tokenCredits } : {}), - } - : {}; - const set = record == null ? syncFields : updateFields; - record = await deps.upsertBalance(user.id, { set, setOnInsert }); + if (record == null) { + // ABSENT record: upsert. The credit rides `$setOnInsert` so a document created + // by a concurrent charge keeps its charged balance; the refill-config sync is + // a legitimate `$set`. + record = await deps.upsertBalance(user.id, { + set: syncFields, + setOnInsert: { + ...(initUser != null ? { user: initUser } : {}), + ...(tokenCredits != null ? { tokenCredits } : {}), + }, + }); + } else if (tokenCredits != null) { + // EXISTING record with a null credit: a blind `$set` would restore credits a + // concurrent initializer/charge already set and spent between the read and + // here. Initialize under a `{ tokenCredits: null }` CAS; on a miss re-read the + // winner's balance instead of overwriting it. + const initialized = await deps.initializeNullBalance(user.id, { + tokenCredits, + sync: syncFields, + }); + record = initialized ?? (await deps.findBalance(user.id)) ?? record; + } else if (Object.keys(syncFields).length > 0) { + // EXISTING record with a real credit: only refill-config sync remains, which + // never touches `tokenCredits` and is safe to `$set` directly. + record = await deps.upsertBalance(user.id, { set: syncFields, setOnInsert: {} }); + } } } const credits = record?.tokenCredits ?? 0; @@ -622,6 +653,18 @@ export function createSchedulesService( return engine; } + /** + * The clustered entrypoint's ONLY schedule maintenance. Exposes the same erasure sweep + * the standard entrypoint falls back to, so a soft-deleted row whose delete/terminal + * erase-on-settle attempts missed still drains instead of retaining the owner's prompt + * forever. It shares startErasureFallback's idempotent startup guard, the sweep's own + * shutdown registration, and the topology-fenced owner-death policy — and arms nothing + * else, so running it in every clustered replica changes nothing about v1 scheduling. + */ + function initializeScheduleErasureSweep(): void { + startErasureFallback(); + } + /** * Manual run-now fire. Acquires the schedule lease to serialize concurrent * run-now clicks (and to block against a background engine claim), then fires @@ -1396,5 +1439,6 @@ export function createSchedulesService( deleteScheduleForOwner, quiesceUserSchedules, initializeScheduleEngine, + initializeScheduleErasureSweep, }; } diff --git a/packages/data-schemas/src/methods/schedule.methods.spec.ts b/packages/data-schemas/src/methods/schedule.methods.spec.ts index d1bdeb45ab..dd6ee20216 100644 --- a/packages/data-schemas/src/methods/schedule.methods.spec.ts +++ b/packages/data-schemas/src/methods/schedule.methods.spec.ts @@ -1189,6 +1189,121 @@ describe('claim-token fencing (stale worker cannot mutate an edited/deleted sche }); }); +describe('owner edit clears a dead paused card', () => { + const scheduledFor = new Date('2026-07-20T12:00:00Z'); + + /** Projects a `requires_action` card whose run captured revision 0, matching a real + * fire: the reservation stamps the config generation on the run row. */ + async function pauseUnderRevisionZero(): Promise { + const schedule = await methods.createSchedule(scheduleData()); + await methods.insertScheduleRun(runData(schedule, { scheduledFor, configRevision: 0 })); + await methods.recordRunOutcome({ + scheduleId: schedule.id, + scheduledFor, + status: 'requires_action', + conversationId: 'convo-paused', + autoDisableAfterFailures: 3, + }); + const paused = await getSchedule(schedule.id); + expect(paused.lastRun?.status).toBe('requires_action'); + return schedule; + } + + it('a rename drops the pause the stale approval can no longer clear', async () => { + const schedule = await pauseUnderRevisionZero(); + const edited = await methods.updateScheduleById(schedule.id, schedule.user, { + name: 'renamed', + }); + expect(edited?.name).toBe('renamed'); + expect(edited?.configRevision).toBe(1); + expect(edited?.lastRun).toBeUndefined(); + }); + + it('a disabling edit drops the pause (no future occurrence would repair it)', async () => { + const schedule = await pauseUnderRevisionZero(); + const edited = await methods.updateScheduleById(schedule.id, schedule.user, { enabled: false }); + expect(edited?.enabled).toBe(false); + expect(edited?.lastRun).toBeUndefined(); + }); + + it('preserves a terminal success card across an edit', async () => { + const schedule = await methods.createSchedule(scheduleData()); + await methods.insertScheduleRun(runData(schedule, { scheduledFor, configRevision: 0 })); + await methods.recordRunOutcome({ + scheduleId: schedule.id, + scheduledFor, + status: 'success', + conversationId: 'convo-ok', + autoDisableAfterFailures: 3, + }); + const edited = await methods.updateScheduleById(schedule.id, schedule.user, { + name: 'renamed', + }); + expect(edited?.lastRun?.status).toBe('success'); + expect(edited?.lastRun?.conversationId).toBe('convo-ok'); + }); + + /** A resume terminalizes the paused occurrence between the handler's read and the + * edit; the clear is fenced on the card STILL being the pause, so the settled result + * survives rather than being cleared on a stale pre-read. */ + it('preserves a terminal outcome that races the edit ahead of it', async () => { + const schedule = await methods.createSchedule(scheduleData()); + await methods.insertScheduleRun(runData(schedule, { scheduledFor, configRevision: 0 })); + await methods.recordRunOutcome({ + scheduleId: schedule.id, + scheduledFor, + status: 'requires_action', + conversationId: 'convo-1', + autoDisableAfterFailures: 3, + }); + // The pause settles to interrupted before the edit lands. + await methods.recordRunOutcome({ + scheduleId: schedule.id, + scheduledFor, + status: 'interrupted', + error: 'stopped', + autoDisableAfterFailures: 3, + }); + const edited = await methods.updateScheduleById(schedule.id, schedule.user, { + name: 'renamed', + }); + expect(edited?.lastRun?.status).toBe('interrupted'); + }); + + it('a stale-revision pause cannot resurrect the cleared card', async () => { + const schedule = await pauseUnderRevisionZero(); + await methods.updateScheduleById(schedule.id, schedule.user, { name: 'renamed' }); + // The run row is still `requires_action` under revision 0; a reconciliation re-affirm + // of the pause must not re-stamp the card now that the schedule advanced to revision 1. + await methods.recordRunOutcome({ + scheduleId: schedule.id, + scheduledFor, + status: 'requires_action', + conversationId: 'convo-paused', + autoDisableAfterFailures: 3, + }); + const after = await getSchedule(schedule.id); + expect(after.lastRun).toBeUndefined(); + }); + + it('a stale old-revision terminal outcome cannot overwrite the edited card or counters', async () => { + const schedule = await pauseUnderRevisionZero(); + await methods.updateScheduleById(schedule.id, schedule.user, { name: 'renamed' }); + // The dead occurrence (revision 0) finally settles; its bookkeeping is revision-fenced, + // so it neither re-projects the card nor advances counters on the revision-1 schedule. + await methods.recordRunOutcome({ + scheduleId: schedule.id, + scheduledFor, + status: 'success', + conversationId: 'convo-paused', + autoDisableAfterFailures: 3, + }); + const after = await getSchedule(schedule.id); + expect(after.lastRun).toBeUndefined(); + expect(after.runCount).toBe(0); + }); +}); + describe('schedule deletion is retryable', () => { /** * The mark is the FIRST of several teardown steps (read active runs, abort their diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index 14c08a6067..1424c11ede 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -359,30 +359,50 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche * persist state derived from a row the other has already replaced. */ options?: { expectedConfigRevision?: number }, ): Promise { - return Schedule() + const filter = { + id, + user: userId, + deleting: { $ne: true }, + ...(options?.expectedConfigRevision !== undefined + ? { configRevision: options.expectedConfigRevision } + : {}), + }; + const mutation = { + $set: { ...update, claimToken: randomUUID() }, + // The ONLY writer of configRevision: an owner edit moves the config + // generation forward atomically with the claim-token rotation, so a run + // that started under the old config can detect it and skip bookkeeping. + // Worker/policy writes (claim, lease, advance, disable, bookkeeping) never + // bump it, and deletion deliberately does not either — a draining run must + // still be able to record its outcome before erasure. + $inc: { configRevision: 1 }, + }; + // FIRST branch: if the card still shows a pause, drop that projection in the SAME + // atomic write as the revision bump. A `requires_action` card present at edit time + // was projected by a run that captured the pre-edit revision; its terminal outcome + // is revision-fenced (projectLastRun), so once this bump lands it can never replace + // the card — and a disabling edit means no future occurrence will either, leaving it + // stuck on "Needs approval" for a dead action. The clear is fenced on the projection + // STILL being that pause (not a stale handler pre-read), so a terminal outcome or a + // newer occurrence that raced in fails this branch and is preserved by the second. + // DocumentDB rules out a conditional pipeline `$unset`, so this is a classic-operator + // CAS branch rather than one update. + const cleared = await Schedule() .findOneAndUpdate( - { - id, - user: userId, - deleting: { $ne: true }, - ...(options?.expectedConfigRevision !== undefined - ? { configRevision: options.expectedConfigRevision } - : {}), - }, - { - $set: { ...update, claimToken: randomUUID() }, - // The ONLY writer of configRevision: an owner edit moves the config - // generation forward atomically with the claim-token rotation, so a run - // that started under the old config can detect it and skip bookkeeping. - // Worker/policy writes (claim, lease, advance, disable, bookkeeping) never - // bump it, and deletion deliberately does not either — a draining run must - // still be able to record its outcome before erasure. - $inc: { configRevision: 1 }, - ...(unset ? { $unset: unset } : {}), - }, + { ...filter, 'lastRun.status': 'requires_action' }, + { ...mutation, $unset: { lastRun: 1, ...(unset ?? {}) } }, { new: true }, ) .lean(); + if (cleared != null) { + return cleared; + } + // SECOND branch: no dead pause to clear (terminal history survives untouched), or the + // row is gone / a concurrent edit moved the revision — a null here is the true + // gone/conflict signal the caller distinguishes. + return Schedule() + .findOneAndUpdate(filter, { ...mutation, ...(unset ? { $unset: unset } : {}) }, { new: true }) + .lean(); } async function deleteScheduleById(id: string, userId: string | Types.ObjectId): Promise {