From 33953429cc6dcbdbbbf3e37a89f98f7ab203f9fc Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Mon, 17 Aug 2026 12:32:22 -0400 Subject: [PATCH] fix: re-fence scheduled resume and expiry --- .../agents/__tests__/resume.spec.js | 5 +- api/server/controllers/agents/resume.js | 5 +- api/server/experimental.js | 1 + api/server/experimental.spec.js | 10 ++++ packages/api/src/schedules/FOLLOWUPS.md | 2 - packages/api/src/schedules/service.spec.ts | 59 ++++++++++++++++++- packages/api/src/schedules/service.ts | 22 ++++++- 7 files changed, 96 insertions(+), 8 deletions(-) diff --git a/api/server/controllers/agents/__tests__/resume.spec.js b/api/server/controllers/agents/__tests__/resume.spec.js index 2fd4d1df54..fd6fa4eeb6 100644 --- a/api/server/controllers/agents/__tests__/resume.spec.js +++ b/api/server/controllers/agents/__tests__/resume.spec.js @@ -389,7 +389,10 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => { expect(res.status).toBe(200); await settled; - expect(mockClaimScheduleResume).toHaveBeenCalledWith('schedule-1', scheduledFor); + expect(mockClaimScheduleResume).toHaveBeenCalledWith('schedule-1', scheduledFor, { + expectedConfigRevision: 4, + automatic: true, + }); expect(mockRecordScheduleOutcome).toHaveBeenCalledWith({ scheduleId: 'schedule-1', diff --git a/api/server/controllers/agents/resume.js b/api/server/controllers/agents/resume.js index 95f7dca9b3..e561a33d10 100644 --- a/api/server/controllers/agents/resume.js +++ b/api/server/controllers/agents/resume.js @@ -810,7 +810,10 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) if (scheduleId) { let scheduleClaim; try { - scheduleClaim = await claimScheduleResume(scheduleId, scheduledFor); + scheduleClaim = await claimScheduleResume(scheduleId, scheduledFor, { + expectedConfigRevision: job.metadata?.scheduleConfigRevision, + automatic: job.metadata?.scheduleManual !== true, + }); } catch (err) { await decrementPendingRequest(userId); logger.error('[ResumeAgentController] Failed to claim scheduled resume capacity', err); diff --git a/api/server/experimental.js b/api/server/experimental.js index 4dd5c94c17..37c67f78eb 100644 --- a/api/server/experimental.js +++ b/api/server/experimental.js @@ -280,6 +280,7 @@ if (cluster.isMaster) { // but an already-fired scheduled generation can still reach HITL here. Settle // its durable run when the generic approval runtime expires it. GenerationJobManager.setApprovalExpiredHandler(recordExpiredScheduleApproval); + GenerationJobManager.initialize(); /** * The master may assign the sweep worker before or after this worker has * loaded app config. These flags join the IPC assignment with config diff --git a/api/server/experimental.spec.js b/api/server/experimental.spec.js index c4364fb6b8..bc6047c921 100644 --- a/api/server/experimental.spec.js +++ b/api/server/experimental.spec.js @@ -23,6 +23,16 @@ describe('Experimental server configuration', () => { expect(source).toMatch(/if \(shuttingDown\) \{[\s\S]*?return;[\s\S]*?Starting a new worker/); }); + it('starts approval expiry after installing the scheduled-run callback', () => { + const handlerIndex = source.indexOf( + 'GenerationJobManager.setApprovalExpiredHandler(recordExpiredScheduleApproval);', + ); + const initializeIndex = source.indexOf('GenerationJobManager.initialize();'); + + expect(handlerIndex).toBeGreaterThan(-1); + expect(initializeIndex).toBeGreaterThan(handlerIndex); + }); + it('runs cross-tenant startup work in the system context', () => { expect(source).toContain('await runAsSystem(seedDatabase);'); expect(source).toMatch( diff --git a/packages/api/src/schedules/FOLLOWUPS.md b/packages/api/src/schedules/FOLLOWUPS.md index d5643e083f..4ebeee3368 100644 --- a/packages/api/src/schedules/FOLLOWUPS.md +++ b/packages/api/src/schedules/FOLLOWUPS.md @@ -22,8 +22,6 @@ Scheduled chats remain experimental and default-off. - The standard server supports multi-replica scheduling only with a confirmed Redis-backed `GenerationJobManager`; in-memory mode requires `SCHEDULES_SINGLE_PROCESS`. The legacy `experimental.js` clustered entrypoint does not arm the engine and rejects schedule writes. -- Scheduled HITL resumes are durably fenced, but user-driven resumes do not reserve scheduler - `fireConcurrency` capacity. - Deferred account deletion blocks new remote API admission but cannot yet drain already-admitted OpenAI-compatible or Responses requests; see [#14594](https://github.com/danny-avila/LibreChat/issues/14594). diff --git a/packages/api/src/schedules/service.spec.ts b/packages/api/src/schedules/service.spec.ts index e1e961e9c8..5a26e92a47 100644 --- a/packages/api/src/schedules/service.spec.ts +++ b/packages/api/src/schedules/service.spec.ts @@ -401,6 +401,9 @@ describe('deleteScheduleForOwner', () => { // The run row is still active, so the erase declines; settlement (and the // erase-on-settle it triggers) belongs to the aborted generation's outcome write. methods.eraseScheduleIfDrained = jest.fn(async () => false); + // Public reads intentionally hide deleting rows. That absence must not turn a + // declined erasure into a false `deleted` response. + methods.getScheduleById.mockResolvedValue(null); await expect(service.deleteScheduleForOwner('s1', 'user-1')).resolves.toBe('draining'); }); @@ -900,7 +903,13 @@ describe('global kill switch', () => { describe('scheduled resume capacity', () => { function makeResumeService(occupancy: { takenSlots: number[]; unslotted: number }) { const methods = { - getScheduleById: jest.fn(async () => ({ id: 's1', user: 'user-1' })), + getScheduleById: jest.fn(async () => ({ + id: 's1', + user: 'user-1', + enabled: true, + configRevision: 3, + })), + getRoleByName: jest.fn(async () => ({ permissions: { SCHEDULES: { USE: true } } })), getCapacityOccupancy: jest.fn(async () => occupancy), markRunResumeClaimed: jest.fn(async (_id, _scheduledFor, capacitySlot) => ({ capacitySlot, @@ -959,6 +968,54 @@ describe('scheduled resume capacity', () => { conflict: 'capacity', }); }); + + it('refuses a resume whose schedule revision changed before the capacity handoff', async () => { + const { service, methods } = makeResumeService({ takenSlots: [], unslotted: 0 }); + + await expect( + service.claimScheduleResume('s1', '2026-08-17T12:00:00.000Z', { + expectedConfigRevision: 2, + automatic: true, + }), + ).resolves.toEqual({ conflict: 'inactive' }); + expect(methods.markRunResumeClaimed).not.toHaveBeenCalled(); + }); + + it('refuses a disabled automatic resume but preserves explicit Run Now semantics', async () => { + const { service, methods } = makeResumeService({ takenSlots: [], unslotted: 0 }); + methods.getScheduleById.mockResolvedValue({ + id: 's1', + user: 'user-1', + enabled: false, + configRevision: 3, + }); + + await expect( + service.claimScheduleResume('s1', '2026-08-17T12:00:00.000Z', { + expectedConfigRevision: 3, + automatic: true, + }), + ).resolves.toEqual({ conflict: 'inactive' }); + await expect( + service.claimScheduleResume('s1', '2026-08-17T12:00:00.000Z', { + expectedConfigRevision: 3, + automatic: false, + }), + ).resolves.toEqual({ capacitySlot: 0 }); + }); + + it('refuses a resume when the owner lost schedule access before the capacity handoff', async () => { + const { service, methods } = makeResumeService({ takenSlots: [], unslotted: 0 }); + methods.getRoleByName.mockResolvedValue({ permissions: { SCHEDULES: { USE: false } } }); + + await expect( + service.claimScheduleResume('s1', '2026-08-17T12:00:00.000Z', { + expectedConfigRevision: 3, + automatic: true, + }), + ).resolves.toEqual({ conflict: 'inactive' }); + expect(methods.markRunResumeClaimed).not.toHaveBeenCalled(); + }); }); describe('deployment-wide limits', () => { diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index 2e508d6f41..a7ae234fc0 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -164,6 +164,7 @@ export interface SchedulesService { claimScheduleResume: ( scheduleId: string, scheduledFor: string | Date, + options?: { expectedConfigRevision?: number; automatic?: boolean }, ) => Promise; /** Guarded rollback when the approval CAS did not consume the paused action. */ releaseScheduleResumeClaim: ( @@ -819,22 +820,34 @@ export function createSchedulesService( async function claimScheduleResume( scheduleId: string, scheduledFor: string | Date, + options?: { expectedConfigRevision?: number; automatic?: boolean }, ): Promise { const schedule = await methods.getScheduleById(scheduleId); if (schedule == null) { return { conflict: 'inactive' }; } + if (options?.automatic !== false && schedule.enabled === false) { + return { conflict: 'inactive' }; + } + if ( + options?.expectedConfigRevision != null && + typeof schedule.configRevision === 'number' && + schedule.configRevision !== options.expectedConfigRevision + ) { + return { conflict: 'inactive' }; + } const owner = await engineDeps.getUserContext(schedule.user); if (owner == null) { return { conflict: 'inactive' }; } return engineDeps.runInTenantContext(owner, async () => { - const [ownerLimits, deploymentLimits, globallyDisabled] = await Promise.all([ + const [ownerLimits, deploymentLimits, globallyDisabled, hasAccess] = await Promise.all([ getLimits(owner), getLimits(), engineDeps.isGloballyDisabled(), + engineDeps.hasScheduleAccess(owner), ]); - if (!ownerLimits.enabled || globallyDisabled) { + if (!ownerLimits.enabled || globallyDisabled || !hasAccess) { return { conflict: 'inactive' }; } const allocation = await engineDeps.withGlobalCapacitySlot( @@ -1143,7 +1156,10 @@ export function createSchedulesService( if (erased) { return 'deleted'; } - return (await methods.getScheduleById(scheduleId)) == null ? 'deleted' : 'draining'; + // eraseScheduleIfDrained already returns true when the row was concurrently + // erased or is absent. A false result therefore means the deleting row is still + // protected by a live lease/run, or the erase could not be confirmed. + return 'draining'; } /**