diff --git a/api/server/controllers/UserController.js b/api/server/controllers/UserController.js index ff10b96cea..06762b8aae 100644 --- a/api/server/controllers/UserController.js +++ b/api/server/controllers/UserController.js @@ -380,6 +380,7 @@ const deleteUserController = async (req, res) => { await db.deleteAllUserMemories(user.id); await db.deleteUserPrompts(user.id); await db.deleteUserSkills(user.id); + await db.deleteSchedulesByUser(user.id); await deleteUserMcpServers(user.id); await db.deleteActions({ user: user.id }); await db.deleteTokens({ userId: user.id }); diff --git a/api/server/controllers/agents/client.js b/api/server/controllers/agents/client.js index 9bc1cf1fd5..f97410360c 100644 --- a/api/server/controllers/agents/client.js +++ b/api/server/controllers/agents/client.js @@ -1485,7 +1485,12 @@ class AgentClient extends BaseClient { // teardown (request.js pause branch / resume.js finally) that would otherwise // release it, and `/resume` 429s under LIMIT_CONCURRENT_MESSAGES. Idempotent via // the flag; if it fails here, the teardown still releases (it checks the flag). - if (!this.pendingRequestReleased) { + // A scheduled fire never acquired an interactive concurrency slot, so it must + // not release one on pause (that would clear a real user's counter). Mark it + // released so downstream teardown skips the decrement too. + if (this.options.req?._isScheduledFire) { + this.pendingRequestReleased = true; + } else if (!this.pendingRequestReleased) { try { await decrementPendingRequest(this.options.req?.user?.id); this.pendingRequestReleased = true; diff --git a/api/server/index.js b/api/server/index.js index 3ba04cd5e8..202a614ef7 100644 --- a/api/server/index.js +++ b/api/server/index.js @@ -356,7 +356,8 @@ const startServer = async () => { // Arm the scheduler only after readiness: an earlier tick could fire // loopback chats at a server that is not yet listening/accepting starts, // recording spurious errors that could auto-disable valid schedules. - initializeScheduleEngine(); + // Ensures indexes before the first tick; failures are logged, not fatal. + await initializeScheduleEngine(); logger.info('Server readiness checks passing.'); } catch (initErr) { serverReady = false; diff --git a/api/server/routes/agents/index.js b/api/server/routes/agents/index.js index 5a13021d3a..35e549ce96 100644 --- a/api/server/routes/agents/index.js +++ b/api/server/routes/agents/index.js @@ -24,6 +24,7 @@ const { messageUserLimiter, } = require('~/server/middleware'); const SteerController = require('~/server/controllers/agents/steer'); +const { recordScheduleOutcome } = require('~/server/services/Schedules'); const { saveMessage } = require('~/models'); const responses = require('./responses'); const openai = require('./openai'); @@ -337,6 +338,17 @@ router.post('/chat/abort', configMiddleware, async (req, res) => { abortResultResponseMessageId: abortResult.jobData?.responseMessageId, }); + // Finalize the schedule side of an aborted scheduled run (incl. a paused one) + // so it doesn't linger as requires_action/started until the abandonment sweep. + if (job.metadata?.scheduleId) { + await recordScheduleOutcome({ + scheduleId: job.metadata.scheduleId, + scheduledFor: job.metadata.scheduledFor, + status: 'interrupted', + conversationId: jobStreamId, + }); + } + // HITL: prune the durable checkpoint of a run aborted while paused, so a new turn // in this conversation can't rehydrate the stale interrupt before the Mongo TTL // reclaims it (thread_id is the stable conversationId). Idempotent / no-op when diff --git a/api/server/services/Schedules/index.js b/api/server/services/Schedules/index.js index 139f810e7d..ab857eec9e 100644 --- a/api/server/services/Schedules/index.js +++ b/api/server/services/Schedules/index.js @@ -5,7 +5,7 @@ const { PermissionBits, PermissionTypes, } = require('librechat-data-provider'); -const { tenantStorage, logger } = require('@librechat/data-schemas'); +const { tenantStorage, runAsSystem, logger } = require('@librechat/data-schemas'); const { checkPermission } = require('~/server/services/PermissionService'); const { fireSchedule, @@ -140,10 +140,18 @@ const engineDeps = { /** @type {ReturnType | undefined} */ let engine; -function initializeScheduleEngine() { +async function initializeScheduleEngine() { if (engine != null) { return engine; } + // Explicitly build the Schedule/ScheduleRun indexes first — the unique + // idempotency index and TTL retention index would otherwise never exist when + // MONGO_AUTO_INDEX is disabled (the production default). + try { + await runAsSystem(() => methods.ensureScheduleIndexes()); + } catch (err) { + logger.error('[schedules] failed to ensure indexes:', err); + } engine = startScheduleEngine(engineDeps); return engine; } diff --git a/client/src/components/SidePanel/Schedules/ScheduleCard.tsx b/client/src/components/SidePanel/Schedules/ScheduleCard.tsx index ec4fd3756d..f864c98e6c 100644 --- a/client/src/components/SidePanel/Schedules/ScheduleCard.tsx +++ b/client/src/components/SidePanel/Schedules/ScheduleCard.tsx @@ -171,7 +171,10 @@ export default function ScheduleCard({ schedule }: ScheduleCardProps) { const lastRunConvoId = schedule.lastRun?.conversationId; return ( -
+
{schedule.name} diff --git a/client/src/components/SidePanel/Schedules/ScheduleDialog.tsx b/client/src/components/SidePanel/Schedules/ScheduleDialog.tsx index a7daf6e352..9cc2371d91 100644 --- a/client/src/components/SidePanel/Schedules/ScheduleDialog.tsx +++ b/client/src/components/SidePanel/Schedules/ScheduleDialog.tsx @@ -115,7 +115,13 @@ export default function ScheduleDialog({ const { showToast } = useToastContext(); const locale = i18n.language; - const { control, register, watch, handleSubmit } = useForm({ + const { + control, + register, + watch, + handleSubmit, + formState: { dirtyFields }, + } = useForm({ defaultValues: getDefaultValues(schedule), }); const name = watch('name'); @@ -199,13 +205,22 @@ export default function ScheduleDialog({ const onSubmit = (values: ScheduleFormValues) => { const cadence = buildCadence(values); if (schedule) { + // Preserve the stored cadence (incl. multi-day weekly `daysOfWeek`, which + // this single-day picker can't represent) unless the user actually touched + // a cadence control — otherwise a rename would silently collapse it. + const cadenceTouched = + dirtyFields.frequency || + dirtyFields.hour12 || + dirtyFields.minute || + dirtyFields.meridiem || + dirtyFields.dayOfWeek; updateSchedule.mutate({ id: schedule.id, payload: { name: values.name.trim(), prompt: values.prompt.trim(), agent_id: values.agent_id, - cadence, + ...(cadenceTouched ? { cadence } : {}), }, }); return; diff --git a/client/src/data-provider/Schedules/queries.ts b/client/src/data-provider/Schedules/queries.ts index 41c1247ac9..a5fd8fee9e 100644 --- a/client/src/data-provider/Schedules/queries.ts +++ b/client/src/data-provider/Schedules/queries.ts @@ -8,8 +8,11 @@ export const useSchedulesQuery = ( config?: UseQueryOptions, ): QueryObserverResult => { return useQuery([QueryKeys.schedules], () => dataService.getSchedules(), { - refetchOnWindowFocus: false, - refetchOnReconnect: false, + // Automatic runs mutate nextRunAt/lastRun/auto-disable server-side while the + // panel is open; refresh on focus and on a modest interval so it stays current. + refetchOnWindowFocus: true, + refetchOnReconnect: true, + refetchInterval: 60_000, ...config, }); }; diff --git a/e2e/specs/mock/schedules.spec.ts b/e2e/specs/mock/schedules.spec.ts new file mode 100644 index 0000000000..af4862aeb2 --- /dev/null +++ b/e2e/specs/mock/schedules.spec.ts @@ -0,0 +1,77 @@ +import { expect, test } from '@playwright/test'; +import type { Page } from '@playwright/test'; + +const uniqueName = (prefix: string) => `${prefix} ${Date.now()}-${Math.floor(Math.random() * 1e4)}`; + +/** + * Seeds a minimal agent via the authenticated API (storageState) so the schedule + * dialog's required agent picker has something to select. Returns the agent name. + */ +async function seedAgent(page: Page): Promise { + const name = uniqueName('E2E Agent'); + const response = await page.request.post('/api/agents', { + data: { name, provider: 'Mock Provider A', model: 'mock-model-a' }, + }); + expect(response.ok()).toBeTruthy(); + const agent = await response.json(); + expect(agent.id).toBeTruthy(); + return name; +} + +async function openSchedulesPanel(page: Page) { + await page.getByTestId('nav-panel-scheduled').click(); + const panel = page.getByRole('region', { name: 'Scheduled chats' }); + await expect(panel).toBeVisible(); + return panel; +} + +test.describe('scheduled chats', () => { + test('creates, persists, toggles, and deletes a schedule', async ({ page }) => { + test.setTimeout(120000); + const agentName = await seedAgent(page); + const scheduleName = uniqueName('E2E Schedule'); + + await page.goto('/c/new', { timeout: 10000 }); + + // Create a schedule through the dialog. + let panel = await openSchedulesPanel(page); + await panel.getByRole('button', { name: 'New schedule' }).click(); + + const dialog = page.getByRole('dialog'); + await dialog.getByLabel('Name').fill(scheduleName); + await dialog.getByLabel('Prompt').fill('Summarize what happened today'); + + // Pick the seeded agent (the picker shows the placeholder until one is chosen). + await dialog.getByText('Select Agent').click(); + await page.getByRole('option', { name: agentName }).click(); + + await dialog.getByRole('button', { name: 'Create', exact: true }).click(); + + // The new schedule renders as a card. + const card = page.getByTestId('schedule-card').filter({ hasText: scheduleName }); + await expect(card).toBeVisible({ timeout: 15000 }); + + // It survives a full reload (persisted through the real backend + DB). + await page.reload(); + panel = await openSchedulesPanel(page); + const reloadedCard = page.getByTestId('schedule-card').filter({ hasText: scheduleName }); + await expect(reloadedCard).toBeVisible({ timeout: 15000 }); + + // Toggling the enabled switch round-trips and re-reads from the server. + const toggle = reloadedCard.getByRole('switch', { name: 'Enabled' }); + await expect(toggle).toBeChecked(); + await toggle.click(); + await expect(toggle).not.toBeChecked(); + + // Delete it via the kebab menu + confirmation dialog. + await reloadedCard.getByRole('button', { name: 'Schedule options' }).click(); + await page.getByRole('menuitem', { name: 'Delete' }).click(); + const confirm = page.getByRole('dialog', { name: /delete schedule/i }); + await confirm.getByRole('button', { name: 'Delete', exact: true }).click(); + + await expect(page.getByTestId('schedule-card').filter({ hasText: scheduleName })).toHaveCount( + 0, + { timeout: 15000 }, + ); + }); +}); diff --git a/packages/api/src/schedules/engine.ts b/packages/api/src/schedules/engine.ts index 2863d14c09..7d01a82f25 100644 --- a/packages/api/src/schedules/engine.ts +++ b/packages/api/src/schedules/engine.ts @@ -44,9 +44,13 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { const jobStatus = run.conversationId ? await deps.getJobStatus(run.conversationId) : null; const ageMs = Date.now() - (run.firedAt?.getTime() ?? 0); // Resolve the run owner's limits so crash-reconciled auto-disable uses - // the same per-principal threshold as an inline completion. + // the same per-principal threshold as an inline completion. Must run in + // the OWNER's tenant context: getLimits resolves config via the ALS + // tenant, and this loop is under runAsSystem (system tenant). const owner = await deps.getUserContext(run.user); - const runLimits = owner ? await deps.getLimits(owner) : limits; + const runLimits = owner + ? await deps.runInTenantContext(owner, () => deps.getLimits(owner)) + : limits; // All transitions go through recordRunOutcome so the schedule's lastRun // (and the card's status chip) tracks the run, including the pause. const finalize = ( @@ -96,6 +100,31 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { } } }); + + // Catch terminal runs whose schedule bookkeeping never landed (a crash + // between the run-row terminalization and the schedule counter update). + const unbookkept = await runAsSystem(() => + deps.methods.getUnbookkeptRuns( + new Date(Date.now() - RECONCILE_MIN_RUN_AGE_MS), + RECONCILE_BATCH, + ), + ); + await runAsSystem(async () => { + for (const run of unbookkept) { + const owner = await deps.getUserContext(run.user); + const runLimits = owner + ? await deps.runInTenantContext(owner, () => deps.getLimits(owner)) + : limits; + await deps.methods.finalizeBookkeeping({ + scheduleId: run.scheduleId, + scheduledFor: run.scheduledFor, + status: run.status as 'success' | 'error' | 'interrupted', + conversationId: run.conversationId, + error: run.error, + autoDisableAfterFailures: runLimits.autoDisableAfterFailures, + }); + } + }); } catch (error) { logger.error('[schedules] run reconciliation failed:', error); } diff --git a/packages/api/src/schedules/fire.ts b/packages/api/src/schedules/fire.ts index c99db5c4b8..65e24c4638 100644 --- a/packages/api/src/schedules/fire.ts +++ b/packages/api/src/schedules/fire.ts @@ -11,6 +11,21 @@ export function buildFireClientRequestId(scheduleId: string, scheduledFor: Date) return `sched:${scheduleId}:${scheduledFor.toISOString()}`; } +/** + * `ambiguous` = the request may already have been accepted and started a billed + * generation (network error / timeout after send). Those must NOT be recorded as + * a definite failure. `ambiguous: false` = the server returned an error response, + * a genuine rejection safe to count. + */ +class ScheduleFireError extends Error { + constructor( + message: string, + readonly ambiguous: boolean, + ) { + super(message); + } +} + async function postChatMessage( deps: ScheduleEngineDeps, schedule: FireableSchedule, @@ -20,8 +35,9 @@ async function postChatMessage( ): Promise<{ conversationId: string }> { const controller = new AbortController(); const timeout = setTimeout(() => controller.abort(), FIRE_REQUEST_TIMEOUT_MS); + let response: Response; try { - const response = await fetch(`${deps.getSelfUrl()}/api/agents/chat/${EModelEndpoint.agents}`, { + response = await fetch(`${deps.getSelfUrl()}/api/agents/chat/${EModelEndpoint.agents}`, { method: 'POST', signal: controller.signal, headers: { @@ -42,18 +58,27 @@ async function postChatMessage( ...(files.length > 0 ? { files } : {}), }), }); - if (!response.ok) { - const body = await response.text().catch(() => ''); - throw new Error(`Fire POST failed (${response.status}): ${body.slice(0, 300)}`); - } - const payload = (await response.json()) as { conversationId?: string }; - if (!payload.conversationId) { - throw new Error('Fire POST returned no conversationId'); - } - return { conversationId: payload.conversationId }; + } catch (error) { + // fetch threw: no response was received. The request may or may not have + // been processed — ambiguous, so don't terminalize as a definite error. + const message = error instanceof Error ? error.message : String(error); + throw new ScheduleFireError(`Fire POST network failure: ${message}`, true); } finally { clearTimeout(timeout); } + if (!response.ok) { + const body = await response.text().catch(() => ''); + // A received error response is a definite rejection (nothing started). + throw new ScheduleFireError( + `Fire POST failed (${response.status}): ${body.slice(0, 300)}`, + false, + ); + } + const payload = (await response.json().catch(() => ({}))) as { conversationId?: string }; + if (!payload.conversationId) { + throw new ScheduleFireError('Fire POST returned no conversationId', true); + } + return { conversationId: payload.conversationId }; } /** @@ -145,6 +170,25 @@ export async function fireSchedule( return { fired: false, skipped: 'balance' as const }; } + // Resolve attachments BEFORE claiming the run row: a transient file-query + // failure here must not orphan a `started` run that consumes capacity. + const requestedFileIds = schedule.file_ids ?? []; + let files: Awaited>; + try { + files = requestedFileIds.length ? await deps.resolveFiles(requestedFileIds, user) : []; + } catch (fileError) { + logger.error( + `[schedules] file resolution failed for ${schedule.id} (will retry):`, + fileError, + ); + // Leave nextRunAt/lease so the next tick retries; no run row was created. + await methods.releaseLease(schedule.id); + return { fired: false, error: 'File resolution failed' }; + } + const droppedFileIds = requestedFileIds.filter( + (id) => !files.some((file) => file.file_id === id), + ); + const run = await methods.insertScheduleRun({ ...baseRun, status: 'started', @@ -155,24 +199,33 @@ export async function fireSchedule( return { fired: false, skipped: 'duplicate' as const }; } - const requestedFileIds = schedule.file_ids ?? []; - const files = requestedFileIds.length ? await deps.resolveFiles(requestedFileIds, user) : []; - const droppedFileIds = requestedFileIds.filter( - (id) => !files.some((file) => file.file_id === id), - ); + // Reserve-then-verify capacity: the insert above is the atomic reservation. + // If the global in-flight count now exceeds the cap, roll it back and retry + // next tick. Never over-admits; may briefly over-reject under contention. + const active = await methods.countActiveRuns(); + if (active > limits.fireConcurrency) { + await methods.deleteScheduleRun(schedule.id, scheduledFor); + await methods.releaseLease(schedule.id); + return { fired: false, skipped: 'capacity' as const }; + } let conversationId: string; try { ({ conversationId } = await postChatMessage(deps, schedule, user.id, scheduledFor, files)); } catch (error) { - // Failure BEFORE the chat was accepted: the generation never started, so - // record the error and count it toward auto-disable. const message = error instanceof Error ? error.message : String(error); - logger.error(`[schedules] fire failed for ${schedule.id}:`, error); + const ambiguous = error instanceof ScheduleFireError && error.ambiguous; + // Ambiguous (network/timeout after send): the generation may have started + // and been billed, so record `interrupted` (no failure count / auto-disable) + // and let reconciliation settle it. Definite rejection: record `error`. + logger.error( + `[schedules] fire ${ambiguous ? 'ambiguously failed' : 'rejected'} for ${schedule.id}:`, + error, + ); await methods.recordRunOutcome({ scheduleId: schedule.id, scheduledFor, - status: 'error', + status: ambiguous ? 'interrupted' : 'error', error: message, autoDisableAfterFailures: ownerLimits.autoDisableAfterFailures, }); diff --git a/packages/api/src/schedules/handlers.ts b/packages/api/src/schedules/handlers.ts index 0fa9fbb4b2..56042b477a 100644 --- a/packages/api/src/schedules/handlers.ts +++ b/packages/api/src/schedules/handlers.ts @@ -177,6 +177,15 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH const cadence = parsed.data.cadence ?? existing.cadence; const timezone = parsed.data.timezone ?? existing.timezone; const enabled = parsed.data.enabled ?? existing.enabled; + // Re-validate the EFFECTIVE (possibly stored) cadence against the current + // floor whenever this edit leaves the schedule enabled — otherwise a bare + // {enabled:true} could re-enable an existing schedule that now runs too often. + if (enabled && cadenceIntervalMinutes(cadence) < limits.minIntervalMinutes) { + res.status(400).json({ + error: `Schedule interval must be at least ${limits.minIntervalMinutes} minutes`, + }); + return; + } const cadenceChanged = parsed.data.cadence != null || parsed.data.timezone != null || parsed.data.enabled != null; const reEnabled = parsed.data.enabled === true && existing.enabled === false; diff --git a/packages/api/src/schedules/types.ts b/packages/api/src/schedules/types.ts index 3673b77fba..25dab9249b 100644 --- a/packages/api/src/schedules/types.ts +++ b/packages/api/src/schedules/types.ts @@ -67,6 +67,7 @@ export interface FireResult { skipped?: | 'overlap' | 'balance' + | 'capacity' | 'duplicate' | 'agent_deleted' | 'user_missing' diff --git a/packages/api/src/stream/implementations/RedisJobStore.ts b/packages/api/src/stream/implementations/RedisJobStore.ts index fb4a5b0aab..a4cb5976e8 100644 --- a/packages/api/src/stream/implementations/RedisJobStore.ts +++ b/packages/api/src/stream/implementations/RedisJobStore.ts @@ -1792,6 +1792,9 @@ export class RedisJobStore implements IJobStore { promptTokens: data.promptTokens ? parseInt(data.promptTokens, 10) : undefined, agent_id: data.agent_id || undefined, isTemporary: data.isTemporary != null ? data.isTemporary === '1' : undefined, + // Scheduled-fire bookkeeping so a Redis-backed HITL resume can record outcome. + scheduleId: data.scheduleId || undefined, + scheduledFor: data.scheduledFor || undefined, // Deferred tools discovered before a HITL pause; replayed into createRun on resume. discoveredTools: data.discoveredTools ? JSON.parse(data.discoveredTools) : undefined, titleEvent: data.titleEvent || undefined, diff --git a/packages/data-schemas/src/methods/schedule.methods.spec.ts b/packages/data-schemas/src/methods/schedule.methods.spec.ts index ae160bc5c5..74d9e403e1 100644 --- a/packages/data-schemas/src/methods/schedule.methods.spec.ts +++ b/packages/data-schemas/src/methods/schedule.methods.spec.ts @@ -480,6 +480,102 @@ describe('recordRunOutcome — reconciled completions and no-match guard', () => }); }); +describe('recordRunOutcome idempotency + crash-retry (bookkeeping)', () => { + const scheduledFor = new Date('2026-07-20T12:00:00Z'); + + it('counts an occurrence at most once across repeated invocations', async () => { + const schedule = await methods.createSchedule(scheduleData()); + await methods.insertScheduleRun(runData(schedule, { scheduledFor })); + const outcome = { + scheduleId: schedule.id, + scheduledFor, + status: 'success' as const, + conversationId: 'convo-1', + autoDisableAfterFailures: 3, + }; + await methods.recordRunOutcome(outcome); + // A second call (e.g. reconciler racing the inline finish) must not re-count. + await methods.recordRunOutcome(outcome); + const updated = await getSchedule(schedule.id); + expect(updated.runCount).toBe(1); + expect((updated as { lastCountedFor?: Date }).lastCountedFor?.toISOString()).toBe( + scheduledFor.toISOString(), + ); + expect((await getRun(schedule.id, scheduledFor)).bookkept).toBe(true); + }); + + it('finalizeBookkeeping recovers a terminalized-but-uncounted run (crash between writes)', async () => { + const schedule = await methods.createSchedule(scheduleData()); + await methods.insertScheduleRun(runData(schedule, { scheduledFor })); + // Simulate a crash after the run row was terminalized but before bookkeeping. + await ScheduleRun.updateOne( + { scheduleId: schedule.id, scheduledFor }, + { $set: { status: 'success', bookkept: false } }, + ); + expect((await getSchedule(schedule.id)).runCount).toBe(0); + + const unbookkept = await methods.getUnbookkeptRuns(new Date(Date.now() + 1000), 100); + expect(unbookkept.map((r) => r.scheduleId)).toContain(schedule.id); + await methods.finalizeBookkeeping({ + scheduleId: schedule.id, + scheduledFor, + status: 'success', + autoDisableAfterFailures: 3, + }); + const updated = await getSchedule(schedule.id); + expect(updated.runCount).toBe(1); + expect((await getRun(schedule.id, scheduledFor)).bookkept).toBe(true); + // No longer surfaced as needing bookkeeping. + expect(await methods.getUnbookkeptRuns(new Date(Date.now() + 1000), 100)).toHaveLength(0); + }); +}); + +describe('getRunsForReconciliation fairness', () => { + it('always includes started runs even behind a backlog of paused rows', async () => { + const s = await methods.createSchedule(scheduleData()); + // Older paused rows + a newer started row; started must still be returned. + for (let i = 0; i < 3; i++) { + await methods.insertScheduleRun( + runData(s, { + scheduledFor: new Date(`2026-07-1${i}T12:00:00Z`), + status: 'requires_action', + firedAt: new Date('2020-01-01T00:00:00Z'), + }), + ); + } + await methods.insertScheduleRun( + runData(s, { + scheduledFor: new Date('2026-07-25T12:00:00Z'), + status: 'started', + firedAt: new Date('2020-06-01T00:00:00Z'), + }), + ); + const runs = await methods.getRunsForReconciliation(new Date('2026-01-01T00:00:00Z'), 2); + expect(runs.some((r) => r.status === 'started')).toBe(true); + }); +}); + +describe('deleteSchedulesByUser', () => { + it('removes the user’s schedules and their runs', async () => { + const userId = new mongoose.Types.ObjectId(); + const a = await methods.createSchedule(scheduleData({ user: userId })); + await methods.createSchedule(scheduleData({ user: userId })); + const other = await methods.createSchedule(scheduleData()); + await methods.insertScheduleRun(runData(a, { scheduledFor: new Date('2026-07-20T12:00:00Z') })); + await methods.insertScheduleRun( + runData(other, { scheduledFor: new Date('2026-07-20T12:00:00Z') }), + ); + + await methods.deleteSchedulesByUser(userId); + + expect(await methods.getSchedulesByUser(userId)).toHaveLength(0); + expect(await getSchedule(other.id)).not.toBeNull(); + // The other user's run survives; the deleted user's runs are gone. + expect(await ScheduleRun.countDocuments({ scheduleId: a.id })).toBe(0); + expect(await ScheduleRun.countDocuments({ scheduleId: other.id })).toBe(1); + }); +}); + describe('acquireManualRunLease / releaseLease', () => { it('serializes concurrent run-now attempts and can be released without advancing', async () => { const schedule = await methods.createSchedule( diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index dff44e0c0a..1a429f4670 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -6,6 +6,7 @@ import type { IScheduleRun, IScheduleRunDocument, } from '~/types/schedule'; +import { createIndexesWithRetry } from '~/utils/retry'; const DUPLICATE_KEY = 11000; @@ -26,6 +27,7 @@ export interface RecordRunOutcomeParams { } export type ScheduleMethods = { + ensureScheduleIndexes: () => Promise; createSchedule: (data: Partial) => Promise; updateScheduleById: ( id: string, @@ -54,6 +56,10 @@ export type ScheduleMethods = { ) => Promise; hasActiveRun: (scheduleId: string) => Promise; countActiveRuns: () => Promise; + deleteScheduleRun: (scheduleId: string, scheduledFor: Date) => Promise; + deleteSchedulesByUser: (userId: string | Types.ObjectId) => Promise; + getUnbookkeptRuns: (olderThan: Date, limit: number) => Promise; + finalizeBookkeeping: (params: RecordRunOutcomeParams) => Promise; recordRunOutcome: (params: RecordRunOutcomeParams) => Promise; recordSkippedRun: ( data: Partial & { @@ -76,6 +82,17 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche const Schedule = () => mongoose.models.Schedule as Model; const ScheduleRun = () => mongoose.models.ScheduleRun as Model; + /** + * Explicitly builds the Schedule/ScheduleRun indexes. Required because the + * standard production setting `MONGO_AUTO_INDEX=` (empty) disables Mongoose's + * automatic index creation — without this the unique idempotency index and the + * TTL retention index would never exist. Called once before the engine starts. + */ + async function ensureScheduleIndexes(): Promise { + await createIndexesWithRetry(Schedule()); + await createIndexesWithRetry(ScheduleRun()); + } + async function createSchedule(data: Partial): Promise { const doc = await Schedule().create(data); return doc.toObject(); @@ -228,14 +245,68 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche return ScheduleRun().countDocuments({ status: 'started' }); } + /** + * Applies the schedule-side bookkeeping (lastRun + counters + auto-disable) for + * a terminal occurrence. Idempotent via the `lastCountedFor` guard: the $inc + * lands at most once per occurrence no matter how many times it is retried + * (inline finish, reconciler catch of an un-`bookkept` run, crash-replay). + */ + async function applyTerminalBookkeeping( + params: RecordRunOutcomeParams & { firedAt: Date }, + ): Promise { + const lastRun = { + conversationId: params.conversationId, + status: params.status, + error: params.error, + firedAt: params.firedAt, + }; + if (params.status === 'interrupted') { + // Not a success or failure — surface on the card, count nothing, but still + // claim the occurrence so a retry doesn't re-run counters. + await Schedule().updateOne( + { id: params.scheduleId, lastCountedFor: { $ne: params.scheduledFor } }, + { $set: { lastRun, lastCountedFor: params.scheduledFor } }, + ); + return; + } + const isFailure = params.status === 'error'; + const schedule = await Schedule() + .findOneAndUpdate( + { id: params.scheduleId, lastCountedFor: { $ne: params.scheduledFor } }, + { + $set: { + lastRun, + lastCountedFor: params.scheduledFor, + ...(isFailure ? {} : { balanceSkipCount: 0 }), + }, + $inc: isFailure ? { failureCount: 1 } : { runCount: 1 }, + ...(isFailure ? {} : { $unset: { disabledReason: 1 } }), + }, + { new: true }, + ) + .lean(); + // schedule == null means this occurrence was already counted (idempotent retry). + if (schedule == null) { + return; + } + if (!isFailure && schedule.failureCount > 0) { + await Schedule().updateOne({ id: params.scheduleId }, { $set: { failureCount: 0 } }); + } + if (isFailure && schedule.failureCount >= params.autoDisableAfterFailures) { + await disableSchedule(params.scheduleId, 'too_many_failures'); + } + } + /** * Terminal (or pause) transition for a run + lastRun/failure bookkeeping. - * Matches a run row still in `started` OR `requires_action` so a run resumed - * from a HITL pause (reconciled `requires_action -> success`) records the - * same lastRun/counter bookkeeping as an inline completion. + * Matches a run row still in `started` OR `requires_action`. Crash-retryable: + * the run row is marked `bookkept:false` at terminalization and only flipped + * to `true` after bookkeeping lands, so a crash in between is re-applied by the + * reconciler (`getUnbookkeptRuns`), while `lastCountedFor` keeps it idempotent. */ async function recordRunOutcome(params: RecordRunOutcomeParams): Promise { const firedAt = new Date(); + const isTerminal = params.status !== 'requires_action'; const matched = await ScheduleRun().updateOne( { scheduleId: params.scheduleId, @@ -245,45 +316,35 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche { $set: { status: params.status, + ...(isTerminal ? { bookkept: false } : {}), ...(params.conversationId ? { conversationId: params.conversationId } : {}), ...(params.error ? { error: params.error } : {}), ...(params.durationMs != null ? { durationMs: params.durationMs } : {}), }, }, ); + // No-match guard: never touch schedule bookkeeping without a matching run + // (protects against a spoofed scheduleId on a normal chat). if ((matched.modifiedCount ?? 0) === 0) { return; } - const lastRun = { - conversationId: params.conversationId, - status: params.status, - error: params.error, - firedAt, - }; - // A pause surfaces on the schedule card (lastRun) but touches no counters, - // so the "Needs approval" chip renders while the run waits. - if (params.status === 'requires_action' || params.status === 'interrupted') { - await Schedule().updateOne({ id: params.scheduleId }, { $set: { lastRun } }); - return; - } - const isFailure = params.status === 'error'; - const schedule = await Schedule() - .findOneAndUpdate( + if (params.status === 'requires_action') { + // Pause surfaces on the card (lastRun) but touches no counters. + await Schedule().updateOne( { id: params.scheduleId }, { - $set: { lastRun, ...(isFailure ? {} : { balanceSkipCount: 0 }) }, - $inc: isFailure ? { failureCount: 1 } : { runCount: 1 }, - ...(isFailure ? {} : { $unset: { disabledReason: 1 } }), + $set: { + lastRun: { conversationId: params.conversationId, status: params.status, firedAt }, + }, }, - { new: true }, - ) - .lean(); - if (!isFailure && schedule != null && schedule.failureCount > 0) { - await Schedule().updateOne({ id: params.scheduleId }, { $set: { failureCount: 0 } }); - } - if (isFailure && schedule != null && schedule.failureCount >= params.autoDisableAfterFailures) { - await disableSchedule(params.scheduleId, 'too_many_failures'); + ); + return; } + await applyTerminalBookkeeping({ ...params, firedAt }); + await ScheduleRun().updateOne( + { scheduleId: params.scheduleId, scheduledFor: params.scheduledFor }, + { $set: { bookkept: true } }, + ); } async function recordSkippedRun( @@ -332,14 +393,65 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche ); } - /** Non-terminal runs old enough to need a job-store status check. */ + /** + * Non-terminal runs old enough to need a job-store status check. Fetches + * `started` (capacity-consuming) and `requires_action` (paused) in separate + * budgeted, firedAt-ordered buckets so a backlog of long-lived paused rows + * can't starve orphaned `started` runs out of every sweep. + */ async function getRunsForReconciliation(olderThan: Date, limit: number): Promise { + const [started, paused] = await Promise.all([ + ScheduleRun() + .find({ status: 'started', firedAt: { $lt: olderThan } }) + .sort({ firedAt: 1 }) + .limit(limit) + .lean(), + ScheduleRun() + .find({ status: 'requires_action', firedAt: { $lt: olderThan } }) + .sort({ firedAt: 1 }) + .limit(limit) + .lean(), + ]); + return [...started, ...paused]; + } + + /** Terminal runs whose schedule bookkeeping never landed (crash between the two writes). */ + async function getUnbookkeptRuns(olderThan: Date, limit: number): Promise { return ScheduleRun() - .find({ status: { $in: ['started', 'requires_action'] }, firedAt: { $lt: olderThan } }) + .find({ + status: { $in: ['success', 'error', 'interrupted'] }, + bookkept: false, + firedAt: { $lt: olderThan }, + }) + .sort({ firedAt: 1 }) .limit(limit) .lean(); } + /** Re-applies (idempotent) bookkeeping for a terminal run and marks it bookkept. */ + async function finalizeBookkeeping(params: RecordRunOutcomeParams): Promise { + await applyTerminalBookkeeping({ ...params, firedAt: new Date() }); + await ScheduleRun().updateOne( + { scheduleId: params.scheduleId, scheduledFor: params.scheduledFor }, + { $set: { bookkept: true } }, + ); + } + + /** Deletes a run row (used to roll back a capacity reservation). */ + async function deleteScheduleRun(scheduleId: string, scheduledFor: Date): Promise { + await ScheduleRun().deleteOne({ scheduleId, scheduledFor }); + } + + /** Cascade for account deletion: removes a user's schedules and their runs. */ + async function deleteSchedulesByUser(userId: string | Types.ObjectId): Promise { + const schedules = await Schedule().find({ user: userId }).select('id').lean<{ id: string }[]>(); + const ids = schedules.map((s) => s.id); + await Schedule().deleteMany({ user: userId }); + if (ids.length > 0) { + await ScheduleRun().deleteMany({ scheduleId: { $in: ids } }); + } + } + async function transitionRunStatus( scheduleId: string, scheduledFor: Date, @@ -354,6 +466,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche } return { + ensureScheduleIndexes, createSchedule, updateScheduleById, deleteScheduleById, @@ -369,6 +482,10 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche setRunFireDetails, hasActiveRun, countActiveRuns, + deleteScheduleRun, + deleteSchedulesByUser, + getUnbookkeptRuns, + finalizeBookkeeping, recordRunOutcome, recordSkippedRun, getRunsForReconciliation, diff --git a/packages/data-schemas/src/schema/schedule.ts b/packages/data-schemas/src/schema/schedule.ts index 37bae99b19..d2ceae7964 100644 --- a/packages/data-schemas/src/schema/schedule.ts +++ b/packages/data-schemas/src/schema/schedule.ts @@ -102,6 +102,10 @@ const scheduleSchema: Schema = new Schema( default: 0, min: 0, }, + /** Guards `recordRunOutcome` counter increments against double-counting on retry. */ + lastCountedFor: { + type: Date, + }, failureCount: { type: Number, default: 0, diff --git a/packages/data-schemas/src/schema/scheduleRun.ts b/packages/data-schemas/src/schema/scheduleRun.ts index 0afb2a457e..0e35ee2eef 100644 --- a/packages/data-schemas/src/schema/scheduleRun.ts +++ b/packages/data-schemas/src/schema/scheduleRun.ts @@ -55,6 +55,10 @@ const scheduleRunSchema: Schema = new Schema( type: Number, min: 0, }, + /** False on a terminal run whose schedule bookkeeping hasn't landed yet (crash-retry marker). */ + bookkept: { + type: Boolean, + }, }, { timestamps: true, @@ -63,5 +67,8 @@ const scheduleRunSchema: Schema = new Schema( scheduleRunSchema.index({ scheduleId: 1, scheduledFor: 1 }, { unique: true }); scheduleRunSchema.index({ scheduleId: 1, firedAt: -1 }); +// Reconciliation sweeps by status; keeps `started` (capacity) fetch cheap and +// prevents long-lived `requires_action` rows from starving the scan. +scheduleRunSchema.index({ status: 1, firedAt: 1 }); export default scheduleRunSchema; diff --git a/packages/data-schemas/src/types/schedule.ts b/packages/data-schemas/src/types/schedule.ts index d15378e7c2..24c996bf55 100644 --- a/packages/data-schemas/src/types/schedule.ts +++ b/packages/data-schemas/src/types/schedule.ts @@ -32,6 +32,7 @@ export interface ISchedule { runCount: number; failureCount: number; balanceSkipCount: number; + lastCountedFor?: Date; createdAt?: Date; updatedAt?: Date; } @@ -51,6 +52,7 @@ export interface IScheduleRun { error?: string; droppedFileIds?: string[]; durationMs?: number; + bookkept?: boolean; createdAt?: Date; updatedAt?: Date; }