CERBERUS/LibreChat
1
0
Fork 0
mirror of https://github.com/danny-avila/LibreChat.git synced 2026-10-22 10:02:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

fix: harden scheduled run lifecycle

Browse source
This commit is contained in:
Danny Avila 2026-08-17 12:00:38 -04:00
parent 65f6e14c14
commit eb1475b4b3
20 changed files with 822 additions and 38 deletions
Download patch file Download diff file Expand all files Collapse all files

56
api/server/controllers/agents/__tests__/resume.spec.js
View file

@ -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);

103
api/server/controllers/agents/resume.js
View file

@ -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,

6
api/server/experimental.js
View file

@ -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

3
api/server/index.js
View file

@ -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(

24
api/server/services/Schedules/index.js
View file

@ -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));

60
api/server/services/Schedules/index.spec.js Normal file
View file

@ -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');
});
});

65
packages/api/src/schedules/engine.spec.ts
View file

@ -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()]);

18
packages/api/src/schedules/engine.ts
View file

@ -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;

81
packages/api/src/schedules/erasure.spec.ts Normal file
View file

@ -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> = {}): 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' }),
);
});
});

13
packages/api/src/schedules/erasure.ts
View file

@ -33,6 +33,9 @@ export interface ScheduleErasureDeps {
>;
/** Job state at a run's conversationId; null = confirmed absent, throw = unknown. */
getJobStatus: (conversationId: string) => Promise<JobState | null>;
/** 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) {

58
packages/api/src/schedules/service.spec.ts
View file

@ -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<Promise<ActiveRun[]>, [string]>().mockResolvedValue([]);

91
packages/api/src/schedules/service.ts
View file

@ -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<FireResult | null>;
recordScheduleOutcome: (input: RecordScheduleOutcomeInput) => Promise<boolean>;
/** 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<ScheduleResumeClaimResult>;
/** Guarded rollback when the approval CAS did not consume the paused action. */
releaseScheduleResumeClaim: (
scheduleId: string,
scheduledFor: string | Date,
capacitySlot: number,
) => Promise<boolean>;
/**
* 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<ScheduleResumeClaimResult> {
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<boolean> {
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,

22
packages/api/src/stream/ApprovalLifecycle.ts
View file

@ -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<number | null> {
): Promise<SerializableJobData | null> {
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;
}
}

57
packages/api/src/stream/GenerationJobManager.ts
View file

@ -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<void>;
export interface CreateGenerationJobOptions {
startupTelemetry?: AgentStartupTelemetry;
initialMetadata?: Partial<t.GenerationJobMetadata>;
@ -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<void> {
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<void> {
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<string, SerializableJobData>();
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;

50
packages/api/src/stream/__tests__/pendingAction.spec.ts
View file

@ -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<void> }
).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);

10
packages/api/src/stream/implementations/InMemoryJobStore.ts
View file

@ -1008,6 +1008,16 @@ export class InMemoryJobStore implements IJobStoreV2 {
return running;
}
async getRequiresActionJobs(): Promise<SerializableJobData[]> {
const paused: SerializableJobData[] = [];
for (const job of this.jobs.values()) {
if (job.status === 'requires_action') {
paused.push(job);
}
}
return paused;
}
async cleanup(): Promise<number> {
const now = Date.now();
const toDelete: Array<{ streamId: string; createdAt: number }> = [];

11
packages/api/src/stream/implementations/RedisJobStore.ts
View file

@ -2671,6 +2671,17 @@ export class RedisJobStore implements IJobStoreV2 {
return jobs;
}
async getRequiresActionJobs(): Promise<SerializableJobData[]> {
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<number> {
const now = Date.now();
const streamIds = await this.redis.smembers(KEYS.runningJobs);

6
packages/api/src/stream/interfaces/IJobStore.ts
View file

@ -678,6 +678,9 @@ export interface IJobStore {
deleteJob(streamId: string, expectedCreatedAt?: number): Promise<boolean>;
hasJob(streamId: string): Promise<boolean>;
getRunningJobs(): Promise<SerializableJobData[]>;
/** 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<SerializableJobData[]>;
cleanup(): Promise<number>;
recordActivity?(streamId: string, expectedCreatedAt?: number): void;
getJobCount(): Promise<number>;
@ -918,6 +921,9 @@ export interface IJobStoreV2 extends IJobStore {
/** Get all running jobs (for cleanup) */
getRunningJobs(): Promise<SerializableJobData[]>;
/** Get durable paused jobs so approval expiry is not process-runtime-dependent. */
getRequiresActionJobs?(): Promise<SerializableJobData[]>;
/** Cleanup expired jobs */
cleanup(): Promise<number>;

48
packages/data-schemas/src/methods/schedule.methods.spec.ts
View file

@ -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();

78
packages/data-schemas/src/methods/schedule.ts
View file

@ -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<void>;
createSchedule: (data: Partial<ISchedule>) => Promise<ISchedule>;
@ -194,7 +201,16 @@ export type ScheduleMethods = {
IScheduleRun,
'status' | 'abortRequestedAt' | 'abortSource' | 'abortPersistedAt'
> | null>;
markRunResumeClaimed: (scheduleId: string, scheduledFor: Date) => Promise<void>;
markRunResumeClaimed: (
scheduleId: string,
scheduledFor: Date,
capacitySlot: number,
) => Promise<ResumedRunReservation>;
releaseRunResumeClaim: (
scheduleId: string,
scheduledFor: Date,
capacitySlot: number,
) => Promise<boolean>;
markRunAbortPersisted: (scheduleId: string, scheduledFor: Date) => Promise<void>;
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<void> {
await ScheduleRun().updateOne(
{ scheduleId, scheduledFor, status: 'requires_action' },
{ $set: { resumeClaimedAt: new Date() } },
{ timestamps: false },
);
async function markRunResumeClaimed(
scheduleId: string,
scheduledFor: Date,
capacitySlot: number,
): Promise<ResumedRunReservation> {
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<boolean> {
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<void> {
await ScheduleRun().updateOne(
{ scheduleId, scheduledFor },
@ -1699,6 +1756,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
requestRunAbort,
getScheduleRunAbortState,
markRunResumeClaimed,
releaseRunResumeClaim,
markRunAbortPersisted,
setRunFireDetails,
countActiveRuns,