diff --git a/api/server/controllers/agents/resume.js b/api/server/controllers/agents/resume.js index 3af7809c06..f8357822f6 100644 --- a/api/server/controllers/agents/resume.js +++ b/api/server/controllers/agents/resume.js @@ -999,18 +999,43 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) GenerationJobManager.setContentParts(streamId, client.contentParts, job.createdAt); } - await client.resumeCompletion({ - resumeValue: mapped.resumeValue, - seedContent, - runSteps: resumeState?.runSteps ?? [], - abortController: job.abortController, - // Carry the user's MCP auth so approved MCP tools run with their credentials. - userMCPAuthMap: result.userMCPAuthMap, - // Replay deferred tools discovered before the pause (captured at pause). The rebuilt - // graph passes `messages: []`, so without these an approved deferred tool would be - // absent from the schema-only toolMap and resume would fail with "unknown tool". - discoveredToolNames: job.metadata?.discoveredTools, - }); + // Keep the resume hand-off fence FRESH for the continuation's whole lifetime: the + // claim-time stamp ages out on its staleness bound, so a continuation that runs + // longer than that and then pauses AGAIN would re-enter `requires_action` with an + // expired fence — quiesce could settle it while the re-pause writes were still + // landing. Refreshed at half the staleness bound; cleared by the pause record or + // aged out after a crash. + let fenceRefresh = null; + if (job.metadata?.scheduleId) { + fenceRefresh = setInterval( + () => + markScheduledRunResumeClaimed( + job.metadata.scheduleId, + new Date(job.metadata.scheduledFor), + ).catch(() => undefined), + 5 * 60_000, + ); + fenceRefresh.unref?.(); + } + + try { + await client.resumeCompletion({ + resumeValue: mapped.resumeValue, + seedContent, + runSteps: resumeState?.runSteps ?? [], + abortController: job.abortController, + // Carry the user's MCP auth so approved MCP tools run with their credentials. + userMCPAuthMap: result.userMCPAuthMap, + // Replay deferred tools discovered before the pause (captured at pause). The rebuilt + // graph passes `messages: []`, so without these an approved deferred tool would be + // absent from the schema-only toolMap and resume would fail with "unknown tool". + discoveredToolNames: job.metadata?.discoveredTools, + }); + } finally { + if (fenceRefresh) { + clearInterval(fenceRefresh); + } + } // The model may pause AGAIN (another tool, or a follow-up question). The pending // action is already persisted + emitted; leave the job `requires_action`. diff --git a/api/server/experimental.js b/api/server/experimental.js index 02d6420c90..9415766c9e 100644 --- a/api/server/experimental.js +++ b/api/server/experimental.js @@ -46,6 +46,8 @@ const { getDeletingSchedules, eraseScheduleIfDrained, markEraseAttempted, + getActiveRunsForSchedule, + recordRunOutcome, } = require('~/models'); const { checkMigrations } = require('./services/start/migration'); const { configureGenerationStreams } = require('@librechat/api'); @@ -405,7 +407,26 @@ if (cluster.isMaster) { // from the very list they would use to retry. Idempotent and drain-checked, so // running it in every worker is safe. startScheduleErasureSweep({ - methods: { getDeletingSchedules, eraseScheduleIfDrained, markEraseAttempted }, + methods: { + getDeletingSchedules, + eraseScheduleIfDrained, + markEraseAttempted, + getActiveRunsForSchedule, + recordRunOutcome, + }, + // Job state for the abandoned-run settle: null = confirmed absent, throw = unknown. + getJobStatus: async (conversationId) => { + const jobState = await GenerationJobManager.getJobStore()?.getJob(conversationId); + if (jobState == null) { + return null; + } + return { + status: jobState.status, + scheduleId: jobState.scheduleId, + scheduledFor: jobState.scheduledFor, + createdEventEmitted: jobState.createdEventEmitted === true, + }; + }, }); // Same category of cleanup: finishes account deletions deferred on an unconfirmed // schedule quiesce (the durable barrier refuses authentication, so no client retry diff --git a/api/server/routes/agents/index.js b/api/server/routes/agents/index.js index c0595bc638..8f78126707 100644 --- a/api/server/routes/agents/index.js +++ b/api/server/routes/agents/index.js @@ -467,12 +467,15 @@ router.post('/chat/abort', configMiddleware, async (req, res) => { // means the stop never left this replica: the peer-owned generation keeps // running and billing, and a client retry would land on the terminal-status // branch below, which historically returned without republishing. Re-signal - // once and answer retryable instead of claiming success. Scheduled runs are - // exempt: their abort stamp + reconciler fence own that guarantee, and their - // owner's settlement is the durable acknowledgement. + // once and answer retryable instead of claiming success. SCHEDULED runs too: + // their abort stamp only fences premature settlement — it neither delivers the + // signal nor waits for it, so a 200 here would still leave the peer generating + // until the owner-death fence. The stamp stays unresolved on this exit (an + // undelivered abort must keep fencing the drains), and the retry never + // consults it: the job is already terminal, so the retry skips the live-job + // stamp path and lands on the terminal branch below. if ( abortResult.success && - !scheduledFireIdentity && abortResult.signalDelivered === false && abortResult.signalPublished === false ) { @@ -502,14 +505,32 @@ router.post('/chat/abort', configMiddleware, async (req, res) => { // A swallowed republication failure must not read as success: with the // signal provably still on this replica, the peer-owned generation keeps // running, so the response stays retryable until a publish leaves (or - // this process turns out to own the generation). - if (!scheduledFireIdentity && !resignal.delivered && !resignal.published) { + // this process turns out to own the generation). Scheduled runs included — + // the stamp fences settlement, not delivery. + if (!resignal.delivered && !resignal.published) { res.set('Retry-After', '2'); return res.status(503).json({ error: 'Stop recorded but not yet delivered to the generation. Please retry.', aborted: null, }); } + // Delivered or republished. For a SCHEDULED run this retry never held the + // live-job stamp (the job was already terminal), so resolveStopAttempt + // no-ops — mark the abort persisted DIRECTLY: this route has no stop-side + // persistence pending on this path, and without the mark the generation + // owner's settlement barrier waits its full timeout on the ORIGINAL + // attempt's stamp before deferring to the reconciler. + if (scheduledFireIdentity && !scheduledStopStamped) { + await markScheduledRunAbortPersisted( + scheduledFireIdentity.scheduleId, + scheduledFireIdentity.scheduledFor, + ).catch((err) => + logger.warn( + `[AgentStream] Failed to mark retry abort persisted: ${jobStreamId}`, + err, + ), + ); + } await resolveStopAttempt(); return res.json({ success: true, aborted: jobStreamId }); } diff --git a/packages/api/src/apiKeys/middleware.ts b/packages/api/src/apiKeys/middleware.ts index 712a0aa113..a5e742c07e 100644 --- a/packages/api/src/apiKeys/middleware.ts +++ b/packages/api/src/apiKeys/middleware.ts @@ -10,7 +10,10 @@ export interface ApiKeyAuthDependencies { userId: Types.ObjectId; keyId: Types.ObjectId; } | null>; - findUser: (query: { _id: string | Types.ObjectId }) => Promise; + findUser: ( + query: { _id: string | Types.ObjectId }, + fieldsToSelect?: string, + ) => Promise; } export interface RemoteAgentAccessDependencies { @@ -35,6 +38,10 @@ export interface RemoteAgentAccessRequest extends ApiKeyAuthRequest { agentPermissions?: number; } +/** Methods that cannot persist data; the sequenced barrier recheck is skipped for + * them so read-only API-key traffic pays no extra round trip. */ +const API_KEY_SAFE_METHODS = new Set(['GET', 'HEAD', 'OPTIONS']); + export function createRequireApiKeyAuth(deps: ApiKeyAuthDependencies) { return async ( req: ApiKeyAuthRequest, @@ -104,6 +111,33 @@ export function createRequireApiKeyAuth(deps: ApiKeyAuthDependencies) { }); } + // The read above can be a pre-barrier snapshot returned AFTER the barrier + // committed. A second read SEQUENCED after the first observes any barrier + // that committed before it — scoped to mutating requests, matching the local + // JWT path (only writes can recreate data during the cascade). Fails closed. + if (!API_KEY_SAFE_METHODS.has(req.method)) { + let barrier: { deletionRequestedAt?: Date } | null = null; + try { + barrier = (await deps.findUser({ _id: keyValidation.userId }, 'deletionRequestedAt')) as { + deletionRequestedAt?: Date; + } | null; + } catch { + barrier = null; + } + if (barrier == null || barrier.deletionRequestedAt != null) { + logger.warn( + `[requireApiKeyAuth] Refusing key for ${keyValidation.userId}: deletion barrier raised or unverifiable`, + ); + return res.status(401).json({ + error: { + message: 'Account deletion in progress', + type: 'invalid_request_error', + code: 'invalid_api_key', + }, + }); + } + } + user.id = (user._id as Types.ObjectId).toString(); req.user = user as IUser & { id: string }; req.apiKeyId = keyValidation.keyId; diff --git a/packages/api/src/schedules/engine.ts b/packages/api/src/schedules/engine.ts index e1db289b1d..45f7ca580a 100644 --- a/packages/api/src/schedules/engine.ts +++ b/packages/api/src/schedules/engine.ts @@ -460,19 +460,27 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { timer.unref?.(); }; + let activePass: Promise | null = null; + async function tick() { if (stopped) { return; } - try { - if (ticks % 4 === 0) { - await reconcile(); + const pass = (async () => { + try { + if (ticks % 4 === 0) { + await reconcile(); + } + ticks += 1; + await runTick(); + } catch (error) { + logger.error('[schedules] tick failed:', error); } - ticks += 1; - await runTick(); - } catch (error) { - logger.error('[schedules] tick failed:', error); - } + })().finally(() => { + activePass = null; + }); + activePass = pass; + await pass; scheduleNext(); } @@ -495,14 +503,20 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { // listener is closing and the generation manager has already begun refusing new jobs. // A tick in that window claims a due occurrence, fails its loopback POST against a // server that is shutting down, and books the failure against the schedule — walking a - // healthy schedule toward auto-disable for nothing more than a restart. This NARROWS - // that window rather than closing it: the listener starts closing before pre-drain - // runs, so a tick already in flight can still lose its POST. Occurrences skipped by - // stopping early are simply still due at restart, within the misfire grace. + // healthy schedule toward auto-disable for nothing more than a restart. Stopping the + // timer alone only narrowed the window: a pass ALREADY in flight could still claim a + // due occurrence and lose its loopback POST against the closing listener, so the + // shutdown also AWAITS the active pass — the coordinator's drain then holds the + // listener open until that pass's fire completes or its claim is released. + // Occurrences skipped by stopping early are simply still due at restart, within the + // misfire grace. registerShutdownTask( 'schedule engine', - () => { + async () => { engine.stop(); + if (activePass) { + await activePass; + } }, { phase: 'pre-drain' }, ); diff --git a/packages/api/src/schedules/erasure.ts b/packages/api/src/schedules/erasure.ts index 209c42c1d6..09a9215cba 100644 --- a/packages/api/src/schedules/erasure.ts +++ b/packages/api/src/schedules/erasure.ts @@ -1,10 +1,22 @@ import { logger, runAsSystem } from '@librechat/data-schemas'; -import type { ScheduleMethods } from '@librechat/data-schemas'; +import type { ScheduleMethods, IScheduleRun } from '@librechat/data-schemas'; +import type { JobState } from './types'; +import { hasResumeHandoffInFlight, hasAbortInFlight } from './types'; import { registerShutdownTask } from '~/app/shutdown'; const SWEEP_MS = 5 * 60_000; const SWEEP_JITTER_MS = 30_000; const SWEEP_BATCH = 100; +/** Runs with no readable job older than this are presumed owner-dead (matches the + * engine reconciler's orphan cutoff). */ +const ABANDONED_RUN_AGE_MS = 30 * 60_000; + +/** Terminal job status → the run outcome it proves (mirror of the quiesce map). */ +const TERMINAL_JOB_OUTCOMES: Record = { + complete: 'success', + error: 'error', + aborted: 'interrupted', +}; export interface ScheduleErasureSweep { stop: () => void; @@ -13,8 +25,83 @@ export interface ScheduleErasureSweep { export interface ScheduleErasureDeps { methods: Pick< ScheduleMethods, - 'getDeletingSchedules' | 'eraseScheduleIfDrained' | 'markEraseAttempted' + | 'getDeletingSchedules' + | 'eraseScheduleIfDrained' + | 'markEraseAttempted' + | 'getActiveRunsForSchedule' + | 'recordRunOutcome' >; + /** Job state at a run's conversationId; null = confirmed absent, throw = unknown. */ + getJobStatus: (conversationId: string) => Promise; +} + +/** Whether the observed job still carries THIS occurrence's scheduled identity. */ +function jobMatchesRun(job: JobState | null, run: IScheduleRun): boolean { + if (job == null || job.scheduleId !== run.scheduleId || job.scheduledFor == null) { + return false; + } + return new Date(job.scheduledFor).getTime() === run.scheduledFor.getTime(); +} + +/** + * Settles the abandoned active runs of a DELETING schedule so the erase below can + * proceed. The clustered entrypoint runs no engine reconciler, and the run TTL now + * (correctly) never expires active rows — so a deleting schedule whose generation + * owner died would otherwise retain the run and the owner's prompt indefinitely. + * Same evidence discipline as the quiesce paths: settle only on positive evidence + * (a terminal identity-matched job, or a confirmed-absent job past the owner-death + * cutoff), and defer anything fenced by an in-flight abort or resume hand-off. + */ +async function settleAbandonedRuns(deps: ScheduleErasureDeps, scheduleId: string): Promise { + const runs = await deps.methods.getActiveRunsForSchedule(scheduleId); + const now = Date.now(); + for (const run of runs) { + try { + if (hasAbortInFlight(run, now) || hasResumeHandoffInFlight(run, now)) { + continue; + } + const job = run.conversationId + ? await deps.getJobStatus(run.conversationId).then( + (state) => ({ known: true, state }), + () => ({ known: false, state: null }), + ) + : { known: true, state: null }; + if (!job.known) { + continue; + } + const identity = jobMatchesRun(job.state, run); + if (identity && job.state!.status === 'running') { + continue; + } + if (identity && job.state!.status === 'requires_action') { + // A paused run of a DELETING schedule: its approval can never be consumed, + // but a fresh pause hand-off may still be writing — the started-row gate + // and the resume fence above already deferred those; a settled-state + // paused row is safe to interrupt. + if (run.status === 'started') { + continue; + } + } + const retained = identity ? TERMINAL_JOB_OUTCOMES[job.state!.status] : undefined; + if (retained == null) { + // No terminal evidence: only presume the owner dead past the cutoff. + const age = now - (run.firedAt?.getTime() ?? 0); + if (age < ABANDONED_RUN_AGE_MS) { + continue; + } + } + await deps.methods.recordRunOutcome({ + scheduleId: run.scheduleId, + scheduledFor: run.scheduledFor, + status: retained ?? 'interrupted', + conversationId: run.conversationId, + ...(retained == null ? { error: 'Schedule deleted' } : {}), + autoDisableAfterFailures: Number.MAX_SAFE_INTEGER, + }); + } catch (err) { + logger.warn(`[schedules] abandoned-run settle failed for ${scheduleId}:`, err); + } + } } /** @@ -42,6 +129,9 @@ export function startScheduleErasureSweep(deps: ScheduleErasureDeps): ScheduleEr await runAsSystem(async () => { const deleting = await deps.methods.getDeletingSchedules(SWEEP_BATCH); for (const schedule of deleting) { + await settleAbandonedRuns(deps, schedule.id).catch((err) => { + logger.warn(`[schedules] abandoned-run pass failed for ${schedule.id}:`, err); + }); await deps.methods.eraseScheduleIfDrained(schedule.id).catch((err) => { logger.warn(`[schedules] erasure sweep failed for ${schedule.id}:`, err); }); diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index 12306d09dc..ce37a57339 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -597,7 +597,7 @@ export function createSchedulesService( if (erasureSweep != null || engine != null) { return; } - erasureSweep = startScheduleErasureSweep({ methods }); + erasureSweep = startScheduleErasureSweep({ methods, getJobStatus: engineDeps.getJobStatus }); } async function initializeScheduleEngine(