From eb1475b4b3be0f5d2f972953552db844587bf2cf Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Mon, 17 Aug 2026 12:00:38 -0400 Subject: [PATCH] fix: harden scheduled run lifecycle --- .../agents/__tests__/resume.spec.js | 56 ++++++++++ api/server/controllers/agents/resume.js | 103 +++++++++++++++++- api/server/experimental.js | 6 + api/server/index.js | 3 +- api/server/services/Schedules/index.js | 24 ++++ api/server/services/Schedules/index.spec.js | 60 ++++++++++ packages/api/src/schedules/engine.spec.ts | 65 +++++++++++ packages/api/src/schedules/engine.ts | 18 ++- packages/api/src/schedules/erasure.spec.ts | 81 ++++++++++++++ packages/api/src/schedules/erasure.ts | 13 +++ packages/api/src/schedules/service.spec.ts | 58 +++++++++- packages/api/src/schedules/service.ts | 91 ++++++++++++++-- packages/api/src/stream/ApprovalLifecycle.ts | 22 +++- .../api/src/stream/GenerationJobManager.ts | 57 +++++++++- .../stream/__tests__/pendingAction.spec.ts | 50 +++++++++ .../implementations/InMemoryJobStore.ts | 10 ++ .../stream/implementations/RedisJobStore.ts | 11 ++ .../api/src/stream/interfaces/IJobStore.ts | 6 + .../src/methods/schedule.methods.spec.ts | 48 ++++++++ packages/data-schemas/src/methods/schedule.ts | 78 +++++++++++-- 20 files changed, 822 insertions(+), 38 deletions(-) create mode 100644 api/server/services/Schedules/index.spec.js create mode 100644 packages/api/src/schedules/erasure.spec.ts diff --git a/api/server/controllers/agents/__tests__/resume.spec.js b/api/server/controllers/agents/__tests__/resume.spec.js index 58dd3f6b7a..2fd4d1df54 100644 --- a/api/server/controllers/agents/__tests__/resume.spec.js +++ b/api/server/controllers/agents/__tests__/resume.spec.js @@ -85,6 +85,8 @@ const mockGetMCPRequestContext = jest.fn(); const mockCleanupMCPRequestContextForReq = jest.fn(); const mockRecordScheduleOutcome = jest.fn(); const mockIsScheduleLive = jest.fn(); +const mockClaimScheduleResume = jest.fn(); +const mockReleaseScheduleResumeClaim = jest.fn(); jest.mock('@librechat/data-schemas', () => ({ ...jest.requireActual('@librechat/data-schemas'), @@ -109,6 +111,8 @@ jest.mock('~/models', () => ({ jest.mock('~/server/services/Schedules', () => ({ recordScheduleOutcome: (...args) => mockRecordScheduleOutcome(...args), + claimScheduleResume: (...args) => mockClaimScheduleResume(...args), + releaseScheduleResumeClaim: (...args) => mockReleaseScheduleResumeClaim(...args), isScheduleLive: (...args) => mockIsScheduleLive(...args), })); @@ -265,6 +269,8 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => { mockGenerationJobManager.approvals.finishPausePersistence.mockResolvedValue(true); mockRecordScheduleOutcome.mockResolvedValue(true); mockIsScheduleLive.mockResolvedValue(true); + mockClaimScheduleResume.mockResolvedValue({ capacitySlot: 0 }); + mockReleaseScheduleResumeClaim.mockResolvedValue(true); // `decrementPendingRequest` runs in the controller's `finally` on every // post-ACK path, so resolving on it signals the async continuation is done. @@ -383,6 +389,8 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => { expect(res.status).toBe(200); await settled; + expect(mockClaimScheduleResume).toHaveBeenCalledWith('schedule-1', scheduledFor); + expect(mockRecordScheduleOutcome).toHaveBeenCalledWith({ scheduleId: 'schedule-1', scheduledFor, @@ -396,6 +404,54 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => { ); }); + it('keeps the approval paused when global scheduled-run capacity is full', async () => { + mockGenerationJobManager.getJob.mockResolvedValue(makeScheduledJob()); + mockClaimScheduleResume.mockResolvedValue({ conflict: 'capacity' }); + + const res = await post(approveBody()); + + expect(res.status).toBe(429); + expect(res.headers['retry-after']).toBe('1'); + expect(res.body).toMatchObject({ code: 'SCHEDULE_CAPACITY' }); + expect(mockGenerationJobManager.approvals.resolve).not.toHaveBeenCalled(); + expect(mockDecrementPendingRequest).toHaveBeenCalledWith(USER_ID); + }); + + it('rolls back scheduled capacity when the approval CAS does not consume the action', async () => { + const job = makeScheduledJob(); + mockGenerationJobManager.getJob.mockResolvedValue(job); + mockGenerationJobManager.approvals.resolve.mockResolvedValue(false); + + const res = await post(approveBody()); + + expect(res.status).toBe(409); + expect(mockReleaseScheduleResumeClaim).toHaveBeenCalledWith('schedule-1', scheduledFor, 0); + }); + + it('settles a scheduled continuation stopped during its resumed segment', async () => { + const job = makeScheduledJob(); + mockGenerationJobManager.getJob.mockResolvedValue(job); + mockInitializeClient.mockImplementation(async () => { + job.abortController.abort(); + return { client: makeClient(), userMCPAuthMap: {} }; + }); + + const res = await post(approveBody()); + expect(res.status).toBe(200); + await settled; + await flush(); + + expect(mockRecordScheduleOutcome).toHaveBeenCalledWith({ + scheduleId: 'schedule-1', + scheduledFor, + streamId: CONVO_ID, + jobCreatedAt: 1000, + status: 'interrupted', + conversationId: CONVO_ID, + error: 'Scheduled run was stopped', + }); + }); + it('does not overwrite the terminal winner when failed-resume finalization loses its CAS', async () => { mockGenerationJobManager.getJob.mockResolvedValue(makeScheduledJob()); mockGenerationJobManager.completeJob.mockResolvedValue(false); diff --git a/api/server/controllers/agents/resume.js b/api/server/controllers/agents/resume.js index ddaae85fce..95f7dca9b3 100644 --- a/api/server/controllers/agents/resume.js +++ b/api/server/controllers/agents/resume.js @@ -30,7 +30,12 @@ const { cleanupMCPRequestContextForReq, } = require('~/server/services/MCPRequestContext'); const { saveMessage, getConvo, getMessages } = require('~/models'); -const { recordScheduleOutcome, isScheduleLive } = require('~/server/services/Schedules'); +const { + recordScheduleOutcome, + claimScheduleResume, + releaseScheduleResumeClaim, + isScheduleLive, +} = require('~/server/services/Schedules'); const { GENERATION_PROTOCOL_HEADER, negotiateNewGenerationProtocol, @@ -791,6 +796,90 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) ); } + // Finish the legacy checkpoint snapshot before claiming scheduled capacity. + // It is independent of the approval claim, and holding a deployment-wide slot + // while an indexed saver read stalls would unnecessarily block other schedules + // and lengthen the Mongo-claim -> approval-CAS hand-off window below. + const checkpointGeneration = await checkpointGenerationPromise; + + // A pause frees its scheduled-run capacity slot. Before consuming the approval, + // atomically promote the run row back to `started` and claim a fresh global slot. + // The database's partial unique indexes arbitrate both deployment capacity and a + // concurrent active occurrence of the same schedule. + let scheduleCapacitySlot; + if (scheduleId) { + let scheduleClaim; + try { + scheduleClaim = await claimScheduleResume(scheduleId, scheduledFor); + } catch (err) { + await decrementPendingRequest(userId); + logger.error('[ResumeAgentController] Failed to claim scheduled resume capacity', err); + return sendGenerationJson( + res, + 500, + { error: 'Failed to reserve scheduled-run capacity' }, + generationProtocolVersion, + ); + } + if ('conflict' in scheduleClaim) { + await decrementPendingRequest(userId); + if (scheduleClaim.conflict === 'capacity' || scheduleClaim.conflict === 'overlap') { + res.set('Retry-After', '1'); + return sendGenerationJson( + res, + 429, + { + code: + scheduleClaim.conflict === 'capacity' + ? 'SCHEDULE_CAPACITY' + : 'SCHEDULE_OCCURRENCE_ACTIVE', + error: + scheduleClaim.conflict === 'capacity' + ? 'Scheduled-run capacity is currently full. Please retry.' + : 'Another occurrence of this schedule is still running. Please retry.', + }, + generationProtocolVersion, + ); + } + return sendGenerationJson( + res, + 409, + { + code: + scheduleClaim.conflict === 'inactive' + ? 'SCHEDULE_NO_LONGER_ACTIVE' + : 'SCHEDULE_RUN_NOT_PAUSED', + error: 'This scheduled run can no longer be resumed', + }, + generationProtocolVersion, + ); + } + scheduleCapacitySlot = scheduleClaim.capacitySlot; + } + + /** Release only when the exact generation demonstrably remains paused. If the + * approval CAS reply is ambiguous and the job cannot be read, retaining the slot + * until reconciliation is the safe direction: releasing it could exceed the cap + * while a committed continuation is already running. */ + const rollbackUnconsumedScheduleClaim = async (currentJob) => { + if ( + scheduleId == null || + scheduleCapacitySlot == null || + currentJob?.createdAt !== job.createdAt || + currentJob?.status !== 'requires_action' + ) { + return; + } + try { + await releaseScheduleResumeClaim(scheduleId, scheduledFor, scheduleCapacitySlot); + } catch (rollbackError) { + logger.warn( + '[ResumeAgentController] Failed to release unconsumed scheduled resume capacity', + rollbackError, + ); + } + }; + // Atomically claim the resume. The single winner drives the run; a racing second // submit (double-click, two tabs) gets false and must not re-drive — that would // re-execute tools and double-bill. @@ -800,10 +889,8 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) // would leak the concurrency slot until the counter TTL expires — spuriously 429'ing // the user when they retry the still-paused approval. Release the slot on that path too. let claimed; - let checkpointGeneration; const providerExecutionId = randomUUID(); try { - checkpointGeneration = await checkpointGenerationPromise; /** The CAS that reopens steering must also publish THIS owner's seal * capability. A separate write after status=`running` leaves a window in * which steer/arm requests read the previous replica's capability. */ @@ -819,6 +906,8 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) job.createdAt, ); } catch (err) { + const currentJob = await GenerationJobManager.getJob(streamId).catch(() => null); + await rollbackUnconsumedScheduleClaim(currentJob); await decrementPendingRequest(userId); logger.error('[ResumeAgentController] Failed to claim resume', err); return sendGenerationJson(res, 500, { error: 'Failed to resume' }, generationProtocolVersion); @@ -826,6 +915,7 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) if (!claimed) { await decrementPendingRequest(userId); const currentJob = await GenerationJobManager.getJob(streamId).catch(() => null); + await rollbackUnconsumedScheduleClaim(currentJob); if (currentJob != null && currentJob.createdAt !== job.createdAt) { return sendGenerationJson(res, 409, { code: 'RUN_REPLACED' }, generationProtocolVersion); } @@ -1108,12 +1198,15 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) } // If the user aborted mid-resume, the abort route already emitted the terminal - // event and finalized the job — don't double-save / double-finalize here. + // event and finalized the job — don't double-save / double-finalize here. This + // continuation is nevertheless the scheduled-run owner, so it must settle the + // run row after observing its own abort; the generic Stop route deliberately + // delegates a running generation's settlement to that generation owner. if (job.abortController.signal.aborted) { logger.debug( `[ResumeAgentController] Aborted during resume; abort route finalizes: ${streamId}`, ); - if (scheduleId && pausePersistenceFailureFinalized) { + if (scheduleId) { await recordScheduleOutcome({ scheduleId, scheduledFor, diff --git a/api/server/experimental.js b/api/server/experimental.js index 54c181c6a5..4dd5c94c17 100644 --- a/api/server/experimental.js +++ b/api/server/experimental.js @@ -28,6 +28,7 @@ const { setupGracefulShutdown, configureMessageFilterRegexValidator, configureFileConfigRegexEngine, + GenerationJobManager, } = require('@librechat/api'); const { connectDb, indexSync } = require('~/db'); const initializeOAuthReconnectManager = require('./services/initializeOAuthReconnectManager'); @@ -36,6 +37,7 @@ 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 { jwtLogin, ldapLogin, passportLogin } = require('~/strategies'); const { updateInterfacePermissions: updateInterfacePerms } = require('@librechat/api'); const { @@ -274,6 +276,10 @@ if (cluster.isMaster) { * Each worker runs a full Express server instance */ const app = express(); + // The clustered entrypoint deliberately does not arm the v1 schedule engine, + // 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); /** * 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/index.js b/api/server/index.js index 9fc182b619..0a0e1a4e29 100644 --- a/api/server/index.js +++ b/api/server/index.js @@ -57,7 +57,7 @@ const { capabilityContextMiddleware } = require('./middleware/roles/capabilities const createValidateImageRequest = require('./middleware/validateImageRequest'); const { initializeGitHubSkillSync } = require('./services/Skills/sync'); const { initializeAgentTriggerService } = require('./services/Agents/triggers'); -const { initializeScheduleEngine } = require('./services/Schedules'); +const { initializeScheduleEngine, recordExpiredScheduleApproval } = require('./services/Schedules'); const { jwtLogin, ldapLogin, passportLogin } = require('~/strategies'); const { startExpiredFileSweep } = require('./services/Files/process'); const { checkMigrations } = require('./services/start/migration'); @@ -121,6 +121,7 @@ const configureGenerationStreams = () => { ...streamServices, cleanupOnComplete: !isEnabled(process.env.STREAM_KEEP_COMPLETED_JOBS), }); + GenerationJobManager.setApprovalExpiredHandler(recordExpiredScheduleApproval); GenerationJobManager.initialize(); // Stop active generations and close their SSE streams while the HTTP server drains. registerShutdownTask( diff --git a/api/server/services/Schedules/index.js b/api/server/services/Schedules/index.js index 0d012bc9b3..77456a671b 100644 --- a/api/server/services/Schedules/index.js +++ b/api/server/services/Schedules/index.js @@ -49,14 +49,38 @@ const invoke = (...args) => getService()[method](...args); +/** Host adapter for the generic generation runtime's approval-expiry seam. The + * callback is intentionally idempotent: a replica relay or schedule reconciliation + * may re-drive the same retained terminal evidence after a transient failure. */ +async function recordExpiredScheduleApproval(streamId, job) { + if (!job?.scheduleId || !job?.scheduledFor) { + return; + } + const recorded = await getService().recordScheduleOutcome({ + scheduleId: job.scheduleId, + scheduledFor: job.scheduledFor, + streamId, + jobCreatedAt: job.createdAt, + status: 'interrupted', + conversationId: job.conversationId ?? streamId, + error: 'Approval expired before a decision was made', + }); + if (!recorded) { + throw new Error(`Failed to settle expired scheduled approval ${job.scheduleId}`); + } +} + module.exports = { getLimits: invoke('getLimits'), fireScheduleNow: invoke('fireScheduleNow'), recordScheduleOutcome: invoke('recordScheduleOutcome'), + claimScheduleResume: invoke('claimScheduleResume'), + releaseScheduleResumeClaim: invoke('releaseScheduleResumeClaim'), isScheduleLive: invoke('isScheduleLive'), deleteScheduleForOwner: invoke('deleteScheduleForOwner'), quiesceUserSchedules: invoke('quiesceUserSchedules'), initializeScheduleEngine: invoke('initializeScheduleEngine'), + recordExpiredScheduleApproval, isUserDeleting: async (userId) => { const methods = require('~/models'); return !(await methods.isAgentTriggerPrincipalActive(userId)); diff --git a/api/server/services/Schedules/index.spec.js b/api/server/services/Schedules/index.spec.js new file mode 100644 index 0000000000..194af1ea68 --- /dev/null +++ b/api/server/services/Schedules/index.spec.js @@ -0,0 +1,60 @@ +const mockRecordScheduleOutcome = jest.fn(); +const mockCreateSchedulesService = jest.fn(() => ({ + recordScheduleOutcome: mockRecordScheduleOutcome, +})); + +jest.mock('@librechat/api', () => ({ + createSchedulesService: (...args) => mockCreateSchedulesService(...args), +})); +jest.mock('mongoose', () => ({ models: {} })); +jest.mock('~/server/services/Config/app', () => ({ getAppConfig: jest.fn() })); +jest.mock('~/server/services/Agents/triggers', () => ({ enqueueAgentTrigger: jest.fn() })); +jest.mock('./access', () => ({ resolveAgentFireAccess: jest.fn() })); +jest.mock('~/models', () => ({ isAgentTriggerPrincipalActive: jest.fn() })); + +const { recordExpiredScheduleApproval } = require('./index'); + +describe('recordExpiredScheduleApproval', () => { + beforeEach(() => { + mockRecordScheduleOutcome.mockReset().mockResolvedValue(true); + }); + + it('ignores ordinary generation approvals', async () => { + await expect( + recordExpiredScheduleApproval('conversation-1', { createdAt: 1000 }), + ).resolves.toBeUndefined(); + + expect(mockRecordScheduleOutcome).not.toHaveBeenCalled(); + }); + + it('settles the exact scheduled generation as interrupted', async () => { + await recordExpiredScheduleApproval('conversation-1', { + createdAt: 1000, + conversationId: 'conversation-1', + scheduleId: 'schedule-1', + scheduledFor: '2026-08-17T12:00:00.000Z', + }); + + expect(mockRecordScheduleOutcome).toHaveBeenCalledWith({ + scheduleId: 'schedule-1', + scheduledFor: '2026-08-17T12:00:00.000Z', + streamId: 'conversation-1', + jobCreatedAt: 1000, + status: 'interrupted', + conversationId: 'conversation-1', + error: 'Approval expired before a decision was made', + }); + }); + + it('surfaces a failed durable settlement instead of reporting false success', async () => { + mockRecordScheduleOutcome.mockResolvedValue(false); + + await expect( + recordExpiredScheduleApproval('conversation-1', { + createdAt: 1000, + scheduleId: 'schedule-1', + scheduledFor: '2026-08-17T12:00:00.000Z', + }), + ).rejects.toThrow('Failed to settle expired scheduled approval schedule-1'); + }); +}); diff --git a/packages/api/src/schedules/engine.spec.ts b/packages/api/src/schedules/engine.spec.ts index ed6cffa933..572cdedef7 100644 --- a/packages/api/src/schedules/engine.spec.ts +++ b/packages/api/src/schedules/engine.spec.ts @@ -398,6 +398,53 @@ describe('reconciliation abort fence', () => { }); }); +describe('reconciliation resume hand-off fence', () => { + const scheduledFor = new Date(0); + const resumedRun = (claimAgeMs: number) => ({ + scheduleId: 'sched-1', + scheduledFor, + user: 'u1', + status: 'started', + conversationId: 'c1', + firedAt: new Date(Date.now() - 60 * 60_000), + resumeClaimedAt: new Date(Date.now() - claimAgeMs), + }); + const pausedJob = async () => ({ + status: 'requires_action', + scheduleId: 'sched-1', + scheduledFor: scheduledFor.toISOString(), + }); + + it('does not release freshly reacquired capacity while approval claiming is in flight', async () => { + const methods = makeMethods(makeClaimedSchedule()); + (methods.getRunsForReconciliation as jest.Mock).mockResolvedValue([resumedRun(60_000)]); + + await reconcileOnce(makeDeps(methods, { getJobStatus: pausedJob })); + + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + }); + + it('does not treat a temporarily missing hand-off job as an orphan', async () => { + const methods = makeMethods(makeClaimedSchedule()); + (methods.getRunsForReconciliation as jest.Mock).mockResolvedValue([resumedRun(60_000)]); + + await reconcileOnce(makeDeps(methods, { getJobStatus: async () => null })); + + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + }); + + it('re-pauses a crashed resume after the hand-off fence expires', async () => { + const methods = makeMethods(makeClaimedSchedule()); + (methods.getRunsForReconciliation as jest.Mock).mockResolvedValue([resumedRun(11 * 60_000)]); + + await reconcileOnce(makeDeps(methods, { getJobStatus: pausedJob })); + + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ scheduleId: 'sched-1', status: 'requires_action' }), + ); + }); +}); + describe('reconciliation preserves the intended outcome', () => { const scheduledFor = new Date(0); /** A run whose inline outcome write never landed: old enough to reconcile, still @@ -442,6 +489,24 @@ describe('reconciliation preserves the intended outcome', () => { expect(clearReconciledJob).toHaveBeenCalled(); }); + it('preserves an interrupted owner outcome instead of converting it to success', async () => { + const methods = makeMethods(makeClaimedSchedule()); + (methods.getRunsForReconciliation as jest.Mock).mockResolvedValue([unsettledRun()]); + await tickOnce( + makeDeps(methods, { + getJobStatus: retainedComplete({ + scheduleOutcome: 'interrupted', + scheduleOutcomeError: 'Stopped by owner', + }), + clearReconciledJob: jest.fn(async () => undefined), + }), + ); + + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ status: 'interrupted', error: 'Stopped by owner' }), + ); + }); + it('recovers a balance refusal that claimed an ERROR terminal as skipped_balance', async () => { const methods = makeMethods(makeClaimedSchedule()); (methods.getRunsForReconciliation as jest.Mock).mockResolvedValue([unsettledRun()]); diff --git a/packages/api/src/schedules/engine.ts b/packages/api/src/schedules/engine.ts index 7ce772a2f6..c56800f5f5 100644 --- a/packages/api/src/schedules/engine.ts +++ b/packages/api/src/schedules/engine.ts @@ -3,8 +3,8 @@ import type { IScheduleRun } from '@librechat/data-schemas'; import type { ScheduleEngineDeps, JobState } from './types'; import { isShutdownInProgress, registerShutdownTask } from '~/app/shutdown'; import { fireSchedule, BALANCE_SKIP_DISABLE_THRESHOLD } from './fire'; +import { hasAbortInFlight, hasResumeHandoffInFlight } from './types'; import { computeNextRunAt } from './cadence'; -import { hasAbortInFlight } from './types'; const TICK_MS = 30_000; const TICK_JITTER_MS = 2_000; @@ -43,13 +43,16 @@ function retainedOutcome( jobState: JobState | null, fallback: 'success' | 'error', ): { - status: 'success' | 'error' | 'skipped_balance'; + status: 'success' | 'error' | 'interrupted' | 'skipped_balance'; error?: string; } { const stamped = jobState?.scheduleOutcome; if (stamped === 'skipped_balance') { return { status: 'skipped_balance' }; } + if (stamped === 'interrupted') { + return { status: 'interrupted', error: jobState?.scheduleOutcomeError }; + } if (stamped === 'error' || fallback === 'error') { return { status: 'error', error: jobState?.scheduleOutcomeError ?? 'Run ended in error' }; } @@ -161,6 +164,14 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { (run.status === 'started' || run.status === 'requires_action') && jobStatus === 'requires_action' ) { + // A resumed run claims capacity in Mongo BEFORE the approval CAS flips + // the job back to `running`. During that short hand-off, the old paused + // job is expected and must not re-pause the run (which would release the + // newly claimed capacity slot while the continuation is about to run). + // A stale stamp still falls through so a crashed resume can recover. + if (run.status === 'started' && hasResumeHandoffInFlight(run, Date.now())) { + continue; + } await finalize('requires_action'); continue; } @@ -217,7 +228,8 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { run.status === 'started' && jobStatus == null && ageMs > ORPHAN_RUN_AGE_MS && - !hasAbortInFlight(run, Date.now()) + !hasAbortInFlight(run, Date.now()) && + !hasResumeHandoffInFlight(run, Date.now()) ) { await finalize('interrupted'); continue; diff --git a/packages/api/src/schedules/erasure.spec.ts b/packages/api/src/schedules/erasure.spec.ts new file mode 100644 index 0000000000..8c1c2eb376 --- /dev/null +++ b/packages/api/src/schedules/erasure.spec.ts @@ -0,0 +1,81 @@ +import type { IScheduleRun, ScheduleMethods } from '@librechat/data-schemas'; +import type { JobState } from './types'; +import { startScheduleErasureSweep } from './erasure'; + +const oldRun = (overrides: Partial = {}): IScheduleRun => + ({ + scheduleId: 'schedule-1', + user: 'user-1', + scheduledFor: new Date('2026-08-17T12:00:00.000Z'), + firedAt: new Date(Date.now() - 31 * 60_000), + conversationId: 'conversation-1', + status: 'started', + ...overrides, + }) as unknown as IScheduleRun; + +async function sweepOnce(options: { + canInferOwnerDeathFromMissingJob: boolean; + job?: JobState | null; +}) { + const methods = { + getDeletingSchedules: jest.fn(async () => [{ id: 'schedule-1' }]), + getActiveRunsForSchedule: jest.fn(async () => [oldRun()]), + recordRunOutcome: jest.fn(async () => undefined), + eraseScheduleIfDrained: jest.fn(async () => false), + markEraseAttempted: jest.fn(async () => undefined), + }; + const sweep = startScheduleErasureSweep({ + methods: methods as unknown as ScheduleMethods, + getJobStatus: jest.fn(async () => options.job ?? null), + canInferOwnerDeathFromMissingJob: options.canInferOwnerDeathFromMissingJob, + }); + await jest.advanceTimersByTimeAsync(5 * 60_000); + sweep.stop(); + return methods; +} + +describe('schedule erasure fallback owner-death evidence', () => { + beforeEach(() => { + jest.useFakeTimers(); + jest.spyOn(Math, 'random').mockReturnValue(0); + }); + + afterEach(() => { + jest.useRealTimers(); + jest.restoreAllMocks(); + }); + + it('does not settle a peer-owned run from process-local job absence', async () => { + const methods = await sweepOnce({ canInferOwnerDeathFromMissingJob: false }); + + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + expect(methods.eraseScheduleIfDrained).toHaveBeenCalledWith('schedule-1'); + }); + + it('retains owner-death recovery when job absence is deployment-authoritative', async () => { + const methods = await sweepOnce({ canInferOwnerDeathFromMissingJob: true }); + + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ + scheduleId: 'schedule-1', + status: 'interrupted', + error: 'Schedule deleted', + }), + ); + }); + + it('still trusts an identity-matched terminal job in an unsafe topology', async () => { + const methods = await sweepOnce({ + canInferOwnerDeathFromMissingJob: false, + job: { + status: 'complete', + scheduleId: 'schedule-1', + scheduledFor: '2026-08-17T12:00:00.000Z', + }, + }); + + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ scheduleId: 'schedule-1', status: 'success' }), + ); + }); +}); diff --git a/packages/api/src/schedules/erasure.ts b/packages/api/src/schedules/erasure.ts index 09a9215cba..d252140840 100644 --- a/packages/api/src/schedules/erasure.ts +++ b/packages/api/src/schedules/erasure.ts @@ -33,6 +33,9 @@ export interface ScheduleErasureDeps { >; /** Job state at a run's conversationId; null = confirmed absent, throw = unknown. */ getJobStatus: (conversationId: string) => Promise; + /** Whether absence in this process's job store proves absence deployment-wide. + * False for the process-local fallback whose scheduler refused unsafe topology. */ + canInferOwnerDeathFromMissingJob: boolean; } /** Whether the observed job still carries THIS occurrence's scheduled identity. */ @@ -82,8 +85,18 @@ async function settleAbandonedRuns(deps: ScheduleErasureDeps, scheduleId: string continue; } } + const settledPause = + identity && job.state!.status === 'requires_action' && run.status === 'requires_action'; const retained = identity ? TERMINAL_JOB_OUTCOMES[job.state!.status] : undefined; if (retained == null) { + // In the unsafe-topology fallback, a peer-owned live job is indistinguishable + // from an absent one. Never turn that local absence (or identity mismatch) into + // owner-death evidence and free its globally visible run/capacity slot. + // An identity-matched, durably paused row is positive local evidence, so it + // remains eligible for the existing age-based deleting-schedule cleanup. + if (!settledPause && !deps.canInferOwnerDeathFromMissingJob) { + continue; + } // No terminal evidence: only presume the owner dead past the cutoff. const age = now - (run.firedAt?.getTime() ?? 0); if (age < ABANDONED_RUN_AGE_MS) { diff --git a/packages/api/src/schedules/service.spec.ts b/packages/api/src/schedules/service.spec.ts index 44a0a253f7..3d36547a4f 100644 --- a/packages/api/src/schedules/service.spec.ts +++ b/packages/api/src/schedules/service.spec.ts @@ -746,7 +746,8 @@ describe('quiesceUserSchedules drain wait', () => { it('does not terminalize a paused row whose resumed generation is still running', async () => { jest.useFakeTimers(); - // The resume promotes only the JOB to `running`; the row stays `requires_action`. + // Backward-compatibility case from a pre-capacity resume (or rolling deploy): + // the job is running while the durable row still reads paused. mockJobStore = { getJob: jest.fn(async () => ({ status: 'running', @@ -896,6 +897,61 @@ 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' })), + getCapacityOccupancy: jest.fn(async () => occupancy), + markRunResumeClaimed: jest.fn(async (_id, _scheduledFor, capacitySlot) => ({ + capacitySlot, + })), + releaseRunResumeClaim: jest.fn(async () => true), + }; + const service = createSchedulesService({ + methods: methods as unknown as SchedulesServiceDeps['methods'], + getAppConfig: jest.fn(async () => ({ + interfaceConfig: { + schedules: { + use: true, + maxPerUser: 10, + minIntervalMinutes: 60, + autoDisableAfterFailures: 5, + fireConcurrency: 1, + }, + }, + })), + findUserById: jest.fn(async () => ({ _id: 'user-1', role: 'USER' })), + findBalance: jest.fn(async () => null), + upsertBalance: jest.fn(async () => null), + resolveAgentFireAccess: jest.fn(async () => 'ok' as const), + isUserDeleting: jest.fn(async () => false), + enqueueAgentTrigger: jest.fn(async () => undefined), + } as unknown as SchedulesServiceDeps); + return { service, methods }; + } + + it('promotes a pause through the global slot allocator and releases by exact slot', async () => { + const { service, methods } = makeResumeService({ takenSlots: [], unslotted: 0 }); + const scheduledFor = '2026-08-17T12:00:00.000Z'; + + await expect(service.claimScheduleResume('s1', scheduledFor)).resolves.toEqual({ + capacitySlot: 0, + }); + expect(methods.markRunResumeClaimed).toHaveBeenCalledWith('s1', new Date(scheduledFor), 0); + await expect(service.releaseScheduleResumeClaim('s1', scheduledFor, 0)).resolves.toBe(true); + expect(methods.releaseRunResumeClaim).toHaveBeenCalledWith('s1', new Date(scheduledFor), 0); + }); + + it('leaves the paused row untouched when deployment capacity is full', async () => { + const { service, methods } = makeResumeService({ takenSlots: [0], unslotted: 0 }); + + await expect(service.claimScheduleResume('s1', '2026-08-17T12:00:00.000Z')).resolves.toEqual({ + conflict: 'capacity', + }); + expect(methods.markRunResumeClaimed).not.toHaveBeenCalled(); + }); +}); + describe('deployment-wide limits', () => { const noRuns = () => jest.fn, [string]>().mockResolvedValue([]); diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index 7b7ff9a316..0d790c0d54 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -92,6 +92,10 @@ export interface RecordScheduleOutcomeInput { error?: string; } +export type ScheduleResumeClaimResult = + | { capacitySlot: number } + | { conflict: 'capacity' | 'overlap' | 'not-paused' | 'inactive' }; + /** * Api-side dependencies the schedules service needs injected: model methods, * config/balance access, and the owner-scoped agent access check. Everything @@ -155,6 +159,18 @@ export interface SchedulesService { limits: ScheduleLimits, ) => Promise; recordScheduleOutcome: (input: RecordScheduleOutcomeInput) => Promise; + /** Re-enters a paused occurrence into the DB-enforced global/same-schedule + * capacity set before its approval job is allowed to resume. */ + claimScheduleResume: ( + scheduleId: string, + scheduledFor: string | Date, + ) => Promise; + /** Guarded rollback when the approval CAS did not consume the paused action. */ + releaseScheduleResumeClaim: ( + scheduleId: string, + scheduledFor: string | Date, + capacitySlot: number, + ) => Promise; /** * Whether a schedule is still live (exists and not soft-deleted). The loopback * chat controller calls this right after creating the generation job to re-fence @@ -528,7 +544,14 @@ export function createSchedulesService( if (erasureSweep != null || engine != null) { return; } - erasureSweep = startScheduleErasureSweep({ methods, getJobStatus: engineDeps.getJobStatus }); + erasureSweep = startScheduleErasureSweep({ + methods, + getJobStatus: engineDeps.getJobStatus, + // If topology itself prevented arming, this process's missing job says + // nothing about peer liveness. If only index creation failed, the topology + // proof still holds and the existing owner-death backstop remains valid. + canInferOwnerDeathFromMissingJob: isTopologySafeToArm(), + }); } async function initializeScheduleEngine(): Promise< @@ -793,6 +816,59 @@ export function createSchedulesService( return true; } + async function claimScheduleResume( + scheduleId: string, + scheduledFor: string | Date, + ): Promise { + const schedule = await methods.getScheduleById(scheduleId); + if (schedule == null) { + 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([ + getLimits(owner), + getLimits(), + engineDeps.isGloballyDisabled(), + ]); + if (!ownerLimits.enabled || globallyDisabled) { + return { conflict: 'inactive' }; + } + const allocation = await engineDeps.withGlobalCapacitySlot( + Math.min(ownerLimits.fireConcurrency, deploymentLimits.fireConcurrency), + async (capacitySlot) => { + const attempt = await methods.markRunResumeClaimed( + scheduleId, + new Date(scheduledFor), + capacitySlot, + ); + return 'conflict' in attempt && attempt.conflict === 'slot-taken' + ? 'slot-taken' + : { claimed: attempt }; + }, + ); + if (allocation === 'capacity') { + return { conflict: 'capacity' }; + } + const claimed = allocation.claimed; + if ('conflict' in claimed) { + return { conflict: claimed.conflict }; + } + return { capacitySlot: claimed.capacitySlot }; + }); + } + + function releaseScheduleResumeClaim( + scheduleId: string, + scheduledFor: string | Date, + capacitySlot: number, + ): Promise { + return methods.releaseRunResumeClaim(scheduleId, new Date(scheduledFor), capacitySlot); + } + /** Aborts an active run's loopback job (identity-guarded). Returns whether the * abort was delivered (false when the job wasn't reachable — e.g. a peer worker's * private store, or a transient error). */ @@ -1076,12 +1152,11 @@ export function createSchedulesService( const active = await methods.getActiveRunsForUser(userId); const unconfirmed: string[] = []; for (const run of active) { - // A RESUMED run's row is still `requires_action`: the resume controller promotes - // only the job-store record to `running` and the row is not terminalized until the - // resumed generation finishes. So the row status alone cannot tell a genuinely - // paused run from one actively generating, and the shortcut below would skip - // waiting for a live generation that can still persist messages. Read the live job - // BEFORE aborting, which settles or deletes it and would erase this evidence. + // Current resumes promote the row back to `started` with a capacity slot, but a + // rolling deploy or crash-era row may still be `requires_action` while its job is + // already `running`. So the row status alone cannot prove a genuine pause. Read + // the live job BEFORE aborting, which settles or deletes it and would erase this + // evidence. // UNKNOWN is not ABSENT. A lookup that THREW is evidence of nothing, while one // that succeeded and returned null is positive evidence that no generation holds // this conversation. Collapsing the two (a bare `.catch(() => null)`) would let a @@ -1221,6 +1296,8 @@ export function createSchedulesService( engineDeps, fireScheduleNow, recordScheduleOutcome, + claimScheduleResume, + releaseScheduleResumeClaim, isScheduleLive, deleteScheduleForOwner, quiesceUserSchedules, diff --git a/packages/api/src/stream/ApprovalLifecycle.ts b/packages/api/src/stream/ApprovalLifecycle.ts index 29797a09cf..f8db79b839 100644 --- a/packages/api/src/stream/ApprovalLifecycle.ts +++ b/packages/api/src/stream/ApprovalLifecycle.ts @@ -1,6 +1,11 @@ import { logger } from '@librechat/data-schemas'; import type { Agents } from 'librechat-data-provider'; -import type { IJobStoreV2, JobMetadataPatch, SteerQueueItem } from '~/stream/interfaces/IJobStore'; +import type { + IJobStoreV2, + JobMetadataPatch, + SerializableJobData, + SteerQueueItem, +} from '~/stream/interfaces/IJobStore'; import type { ActivityPhaseSnapshot } from '~/agents/activityPhases/runtime'; import { isPendingActionExpired, @@ -377,15 +382,16 @@ export class ApprovalLifecycle { } /** - * Expires the observed approval and returns the winning job identity. Callers - * that need to notify runtime-local subscribers can use the identity to avoid - * delivering the predecessor's terminal event to a replacement generation. + * Expires the observed approval and returns the winning pre-transition job. + * Callers that need to notify runtime-local subscribers or run host lifecycle + * hooks get both the exact generation identity and its trusted metadata without + * a racy post-transition read (the terminal TTL may already remove the hash). */ async expireWithIdentity( streamId: string, expectedActionId?: string, expectedCreatedAt?: number, - ): Promise { + ): Promise { const job = await this.waitForPausePersistence(streamId, expectedCreatedAt); if ( !job || @@ -394,6 +400,10 @@ export class ApprovalLifecycle { ) { return null; } + // Built-in in-memory stores may mutate the returned job object in place during + // transitionStatus. Snapshot it before the CAS so the host hook receives the + // observed generation metadata even when terminal TTL removes or rewrites it. + const expiredJob = { ...job }; const createdAt = job.createdAt; const ok = await this.store.transitionStatus(streamId, { from: 'requires_action', @@ -409,6 +419,6 @@ export class ApprovalLifecycle { this.callbacks.onExpired?.(streamId, createdAt); logger.debug(`[ApprovalLifecycle] expired pending review: ${streamId}`); } - return ok ? createdAt : null; + return ok ? expiredJob : null; } } diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index 29f1b31c30..eb5ab037b5 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -449,6 +449,13 @@ export interface GenerationJobManagerOptions { cleanupOnComplete?: boolean; } +/** Host-owned lifecycle seam for durable work layered on agent generations. + * Delivery is at-least-once: implementations must be idempotent by generation. */ +export type ApprovalExpiredHandler = ( + streamId: string, + job: SerializableJobData, +) => void | Promise; + export interface CreateGenerationJobOptions { startupTelemetry?: AgentStartupTelemetry; initialMetadata?: Partial; @@ -692,6 +699,10 @@ class GenerationJobManagerClass { /** Whether to cleanup event transport immediately on job completion */ private _cleanupOnComplete = true; + /** Optional host hook; the generic stream runtime does not know what external + * durable work (scheduled chats, webhooks, etc.) a generation represents. */ + private approvalExpiredHandler: ApprovalExpiredHandler | undefined; + constructor(options?: GenerationJobManagerOptions) { const jobStore = options?.jobStore ?? new InMemoryJobStore({ ttlAfterComplete: 0, maxJobs: 1000 }); @@ -820,6 +831,12 @@ class GenerationJobManagerClass { return this._isRedis; } + /** Installs the application-owned approval-expiry hook without coupling the + * stream package to any particular trigger/scheduler implementation. */ + setApprovalExpiredHandler(handler?: ApprovalExpiredHandler): void { + this.approvalExpiredHandler = handler; + } + private get storeLabel(): GenerationJobStore { return this._isRedis ? 'redis' : 'memory'; } @@ -7003,19 +7020,34 @@ class GenerationJobManagerClass { ); } } - const expiredCreatedAt = await this._approvals.expireWithIdentity( + const expiredJob = await this._approvals.expireWithIdentity( streamId, actionId, expectedCreatedAt ?? observedJob?.createdAt, ); - if (expiredCreatedAt == null) { + if (expiredJob == null) { return false; } - await this.notifyApprovalExpiredRuntime(streamId, expiredCreatedAt, observedRuntime); + await this.runApprovalExpiredHandler(streamId, expiredJob); + await this.notifyApprovalExpiredRuntime(streamId, expiredJob.createdAt, observedRuntime); return true; } + private async runApprovalExpiredHandler( + streamId: string, + job: SerializableJobData, + ): Promise { + try { + await this.approvalExpiredHandler?.(streamId, job); + } catch (err) { + // Expiry itself already won its exact CAS. Keep terminal notification moving; + // replica relays may deliver this idempotent hook again, and host-owned durable + // evidence remains available to that host's own recovery path. + logger.error(`[GenerationJobManager] Approval-expiry host hook failed: ${streamId}`, err); + } + } + private async notifyApprovalExpiredRuntime( streamId: string, createdAt: number, @@ -7057,10 +7089,24 @@ class GenerationJobManagerClass { private async expireStaleApprovals(): Promise { let changed = false; - for (const streamId of this.runtimeState.keys()) { + // Scan durable pauses as well as local runtimes. A process can restart while an + // approval waits; if expiry were limited to runtimeState, store cleanup would + // terminalize that ownerless job without crossing the host lifecycle hook. + const candidates = new Map(); + if (this.jobStore.getRequiresActionJobs) { + try { + for (const job of await this.jobStore.getRequiresActionJobs()) { + candidates.set(job.streamId, job); + } + } catch (err) { + logger.error('[GenerationJobManager] Failed to enumerate pending approvals', err); + } + } + const streamIds = new Set([...this.runtimeState.keys(), ...candidates.keys()]); + for (const streamId of streamIds) { let job: SerializableJobData | null; try { - job = await this.jobStore.getJob(streamId); + job = candidates.get(streamId) ?? (await this.jobStore.getJob(streamId)); } catch (err) { logger.error( `[GenerationJobManager] Failed to read job during approval expiry sweep: ${streamId}`, @@ -7077,6 +7123,7 @@ class GenerationJobManagerClass { // The `errorEvent` flag (set by emitError) keeps this idempotent vs the win path. const runtime = this.runtimeState.get(streamId); if (job?.status === 'aborted' && job.error === APPROVAL_EXPIRED_ERROR) { + await this.runApprovalExpiredHandler(streamId, job); await this.notifyApprovalExpiredRuntime(streamId, job.createdAt, runtime); changed = this.releaseJobOwnership(streamId, job.createdAt) || changed; continue; diff --git a/packages/api/src/stream/__tests__/pendingAction.spec.ts b/packages/api/src/stream/__tests__/pendingAction.spec.ts index 551e68f60f..89bb58350e 100644 --- a/packages/api/src/stream/__tests__/pendingAction.spec.ts +++ b/packages/api/src/stream/__tests__/pendingAction.spec.ts @@ -782,6 +782,56 @@ describe('ApprovalLifecycle via GenerationJobManager.approvals (in-memory)', () }); describe('expireApproval notification', () => { + test('delivers the exact expired generation to the host lifecycle hook', async () => { + const streamId = 'stream-expire-host-hook'; + const job = await manager.createJob(streamId, 'user-1'); + await manager.updateMetadata(streamId, { + scheduleId: 'schedule-1', + scheduledFor: '2026-08-17T12:00:00.000Z', + }); + await manager.approvals.pause(streamId, buildAction(streamId)); + const onApprovalExpired = jest.fn(async () => undefined); + manager.setApprovalExpiredHandler(onApprovalExpired); + + expect(await manager.expireApproval(streamId)).toBe(true); + + expect(onApprovalExpired).toHaveBeenCalledWith( + streamId, + expect.objectContaining({ + createdAt: job.createdAt, + status: 'requires_action', + scheduleId: 'schedule-1', + scheduledFor: '2026-08-17T12:00:00.000Z', + }), + ); + }); + + test('expires a durable paused job even when no process-local runtime survived', async () => { + const streamId = 'stream-expire-ownerless-host-hook'; + const job = await jobStore.createJob(streamId, 'user-1', streamId, undefined, { + scheduleId: 'schedule-ownerless', + scheduledFor: '2026-08-17T12:00:00.000Z', + }); + const lifecycle = new ApprovalLifecycle(jobStore); + await lifecycle.pause(streamId, buildAction(streamId, { expiresAt: Date.now() - 1 })); + const onApprovalExpired = jest.fn(async () => undefined); + manager.setApprovalExpiredHandler(onApprovalExpired); + + await ( + manager as unknown as { expireStaleApprovals: () => Promise } + ).expireStaleApprovals(); + + expect(onApprovalExpired).toHaveBeenCalledWith( + streamId, + expect.objectContaining({ + createdAt: job.createdAt, + scheduleId: 'schedule-ownerless', + status: 'requires_action', + }), + ); + await expect(jobStore.getJob(streamId)).resolves.toMatchObject({ status: 'aborted' }); + }); + test('does not expire a replacement paused on the same action id', async () => { const streamId = 'stream-expire-epoch-fence'; const now = jest.spyOn(Date, 'now').mockReturnValue(1000); diff --git a/packages/api/src/stream/implementations/InMemoryJobStore.ts b/packages/api/src/stream/implementations/InMemoryJobStore.ts index 5cf7bbbc26..08b0515e63 100644 --- a/packages/api/src/stream/implementations/InMemoryJobStore.ts +++ b/packages/api/src/stream/implementations/InMemoryJobStore.ts @@ -1008,6 +1008,16 @@ export class InMemoryJobStore implements IJobStoreV2 { return running; } + async getRequiresActionJobs(): Promise { + const paused: SerializableJobData[] = []; + for (const job of this.jobs.values()) { + if (job.status === 'requires_action') { + paused.push(job); + } + } + return paused; + } + async cleanup(): Promise { const now = Date.now(); const toDelete: Array<{ streamId: string; createdAt: number }> = []; diff --git a/packages/api/src/stream/implementations/RedisJobStore.ts b/packages/api/src/stream/implementations/RedisJobStore.ts index 0ed84f1b5d..f7a440e21f 100644 --- a/packages/api/src/stream/implementations/RedisJobStore.ts +++ b/packages/api/src/stream/implementations/RedisJobStore.ts @@ -2671,6 +2671,17 @@ export class RedisJobStore implements IJobStoreV2 { return jobs; } + async getRequiresActionJobs(): Promise { + const streamIds = await this.redis.smembers(KEYS.requiresActionJobs); + if (streamIds.length === 0) { + return []; + } + const jobs = await Promise.all(streamIds.map((streamId) => this.getJob(streamId))); + return jobs.filter( + (job): job is SerializableJobData => job != null && job.status === 'requires_action', + ); + } + async cleanup(): Promise { const now = Date.now(); const streamIds = await this.redis.smembers(KEYS.runningJobs); diff --git a/packages/api/src/stream/interfaces/IJobStore.ts b/packages/api/src/stream/interfaces/IJobStore.ts index d90993b0e4..95525109c7 100644 --- a/packages/api/src/stream/interfaces/IJobStore.ts +++ b/packages/api/src/stream/interfaces/IJobStore.ts @@ -678,6 +678,9 @@ export interface IJobStore { deleteJob(streamId: string, expectedCreatedAt?: number): Promise; hasJob(streamId: string): Promise; getRunningJobs(): Promise; + /** Optional durable paused-job enumeration. Built-in stores implement it so + * the manager can own approval expiry even after the original runtime died. */ + getRequiresActionJobs?(): Promise; cleanup(): Promise; recordActivity?(streamId: string, expectedCreatedAt?: number): void; getJobCount(): Promise; @@ -918,6 +921,9 @@ export interface IJobStoreV2 extends IJobStore { /** Get all running jobs (for cleanup) */ getRunningJobs(): Promise; + /** Get durable paused jobs so approval expiry is not process-runtime-dependent. */ + getRequiresActionJobs?(): Promise; + /** Cleanup expired jobs */ cleanup(): Promise; diff --git a/packages/data-schemas/src/methods/schedule.methods.spec.ts b/packages/data-schemas/src/methods/schedule.methods.spec.ts index 0619858d79..ba2bd21e76 100644 --- a/packages/data-schemas/src/methods/schedule.methods.spec.ts +++ b/packages/data-schemas/src/methods/schedule.methods.spec.ts @@ -301,6 +301,54 @@ describe('reserveStartedRun (single-active overlap guard)', () => { }); }); +describe('markRunResumeClaimed (paused resume capacity)', () => { + const when = new Date('2026-07-20T12:00:00Z'); + + it('atomically promotes a paused run and claims a global capacity slot', async () => { + const schedule = await methods.createSchedule(scheduleData()); + await ScheduleRun.create(runData(schedule, { scheduledFor: when, status: 'requires_action' })); + + await expect(methods.markRunResumeClaimed(schedule.id, when, 0)).resolves.toEqual({ + capacitySlot: 0, + }); + await expect(getRun(schedule.id, when)).resolves.toMatchObject({ + status: 'started', + capacitySlot: 0, + resumeClaimedAt: expect.any(Date), + }); + }); + + it('reports global-slot and same-schedule overlap conflicts without changing the pause', async () => { + const first = await methods.createSchedule(scheduleData()); + const second = await methods.createSchedule(scheduleData()); + await ScheduleRun.create(runData(first, { capacitySlot: 0 })); + await ScheduleRun.create(runData(second, { scheduledFor: when, status: 'requires_action' })); + + await expect(methods.markRunResumeClaimed(second.id, when, 0)).resolves.toEqual({ + conflict: 'slot-taken', + }); + + const later = new Date('2026-07-21T12:00:00Z'); + await ScheduleRun.create(runData(first, { scheduledFor: later, status: 'requires_action' })); + await expect(methods.markRunResumeClaimed(first.id, later, 1)).resolves.toEqual({ + conflict: 'overlap', + }); + await expect(getRun(second.id, when)).resolves.toMatchObject({ status: 'requires_action' }); + await expect(getRun(first.id, later)).resolves.toMatchObject({ status: 'requires_action' }); + }); + + it('rolls back only the exact resume reservation', async () => { + const schedule = await methods.createSchedule(scheduleData()); + await ScheduleRun.create(runData(schedule, { scheduledFor: when, status: 'requires_action' })); + await methods.markRunResumeClaimed(schedule.id, when, 2); + + await expect(methods.releaseRunResumeClaim(schedule.id, when, 1)).resolves.toBe(false); + await expect(methods.releaseRunResumeClaim(schedule.id, when, 2)).resolves.toBe(true); + await expect(getRun(schedule.id, when)).resolves.toMatchObject({ status: 'requires_action' }); + expect((await getRun(schedule.id, when)).capacitySlot).toBeUndefined(); + }); +}); + describe('createScheduleWithSlot (atomic per-user cap)', () => { it('allows exactly maxPerUser and reports limit under concurrency', async () => { const user = new mongoose.Types.ObjectId(); diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index dbbb55d6b3..f0c25e1fe5 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -132,6 +132,13 @@ export type StartedRunReservation = existingStatus?: ScheduleRunStatus; }; +/** Outcome of promoting a paused occurrence back into the capacity-consuming + * `started` set. Both conflicts are database-arbitrated by the same partial + * unique indexes used for a first fire. */ +export type ResumedRunReservation = + | { capacitySlot: number } + | { conflict: 'not-paused' | 'overlap' | 'slot-taken' }; + export type ScheduleMethods = { ensureScheduleIndexes: () => Promise; createSchedule: (data: Partial) => Promise; @@ -194,7 +201,16 @@ export type ScheduleMethods = { IScheduleRun, 'status' | 'abortRequestedAt' | 'abortSource' | 'abortPersistedAt' > | null>; - markRunResumeClaimed: (scheduleId: string, scheduledFor: Date) => Promise; + markRunResumeClaimed: ( + scheduleId: string, + scheduledFor: Date, + capacitySlot: number, + ) => Promise; + releaseRunResumeClaim: ( + scheduleId: string, + scheduledFor: Date, + capacitySlot: number, + ) => Promise; markRunAbortPersisted: (scheduleId: string, scheduledFor: Date) => Promise; setRunFireDetails: ( scheduleId: string, @@ -805,9 +821,6 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche > | null>(); } - /** Stamped by the interactive Stop route once ALL its writes (checkpoint prune, - * partial-response save) have landed; releases the generation owner's settlement - * barrier. Bookkeeping only: never touches `updatedAt`. */ /** * Stamps a paused run as RESUME-CLAIMED: its approval was consumed and a * continuation is running. While fresh, a re-paused job's state is a hand-off in @@ -815,14 +828,58 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche * record ($unset below) is the completion signal; a crash leaves the stamp to * age out on the caller's staleness bound. */ - async function markRunResumeClaimed(scheduleId: string, scheduledFor: Date): Promise { - await ScheduleRun().updateOne( - { scheduleId, scheduledFor, status: 'requires_action' }, - { $set: { resumeClaimedAt: new Date() } }, - { timestamps: false }, - ); + async function markRunResumeClaimed( + scheduleId: string, + scheduledFor: Date, + capacitySlot: number, + ): Promise { + try { + const claimed = await ScheduleRun().findOneAndUpdate( + { scheduleId, scheduledFor, status: 'requires_action' }, + { + $set: { + status: 'started', + capacitySlot, + resumeClaimedAt: new Date(), + }, + }, + { new: true, timestamps: false }, + ); + return claimed == null ? { conflict: 'not-paused' } : { capacitySlot }; + } catch (error) { + if (isCapacitySlotConflict(error)) { + return { conflict: 'slot-taken' }; + } + if (isActiveRunConflict(error)) { + return { conflict: 'overlap' }; + } + throw error; + } } + /** Roll back a resume reservation only while the exact slot this caller claimed + * still belongs to this occurrence. The approval job is checked separately by the + * controller before crossing this seam, so a committed/racing resume never has its + * capacity released by a losing request. */ + async function releaseRunResumeClaim( + scheduleId: string, + scheduledFor: Date, + capacitySlot: number, + ): Promise { + const released = await ScheduleRun().updateOne( + { scheduleId, scheduledFor, status: 'started', capacitySlot }, + { + $set: { status: 'requires_action' }, + $unset: { capacitySlot: 1, resumeClaimedAt: 1 }, + }, + { timestamps: false }, + ); + return (released.modifiedCount ?? 0) > 0; + } + + /** Stamped by the interactive Stop route once ALL its writes (checkpoint prune, + * partial-response save) have landed; releases the generation owner's settlement + * barrier. Bookkeeping only: never touches `updatedAt`. */ async function markRunAbortPersisted(scheduleId: string, scheduledFor: Date): Promise { await ScheduleRun().updateOne( { scheduleId, scheduledFor }, @@ -1699,6 +1756,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche requestRunAbort, getScheduleRunAbortState, markRunResumeClaimed, + releaseRunResumeClaim, markRunAbortPersisted, setRunFireDetails, countActiveRuns,