diff --git a/api/server/controllers/agents/request.js b/api/server/controllers/agents/request.js index e63ea5f31e..f1dde7c8ef 100644 --- a/api/server/controllers/agents/request.js +++ b/api/server/controllers/agents/request.js @@ -977,13 +977,14 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit // in-flight user-message / conversation save, then tear down WITHOUT saving a // partial response, emitting a terminal event, or completing the job. if (client?.pendingApproval) { - // A disconnect-partial save launched while this segment streamed must land - // before the pause is recorded: the run's pause hand-off (and any deletion - // drain that settles a paused run) treats the recorded pause as "all of this - // segment's writes are durable". Never-rejecting by construction. - if (partialSavePromise) { - await partialSavePromise; - } + // Every write launched during this segment must land before the pause is + // recorded: a deletion drain treats a recorded pause as settleable and can + // cascade immediately after. That covers the disconnect-partial save, the + // background user-message save, AND the immediate-mode title — a title is + // billed work (balance upsert + transaction insert), and its aborted task + // unwinding through usage persistence after the cascade would recreate rows + // for a deleted account. All never-rejecting by construction. + await awaitPendingPersistence(); if (response?.databasePromise) { try { await response.databasePromise; diff --git a/packages/api/src/schedules/handlers.spec.ts b/packages/api/src/schedules/handlers.spec.ts index 8a8fe523b3..314304e72c 100644 --- a/packages/api/src/schedules/handlers.spec.ts +++ b/packages/api/src/schedules/handlers.spec.ts @@ -538,12 +538,13 @@ describe('create idempotency', () => { }); }); -describe('deferred erase retry', () => { - it('re-drives the erase of the caller’s soft-deleted schedules on list', async () => { - const deps = makeCreateDeps(); +describe('deferred deletion retry', () => { + it('re-drives the full deletion of the caller’s soft-deleted schedules on list', async () => { + const deps = makeCreateDeps({ + deleteSchedule: jest.fn(async () => 'deleted'), + } as Partial); (deps.methods.getSchedulesByUser as jest.Mock) = jest.fn(async () => []); (deps.methods.getDeletingScheduleIds as jest.Mock) = jest.fn(async () => ['stranded-1']); - (deps.methods.eraseScheduleIfDrained as jest.Mock) = jest.fn(async () => true); const { res } = makeRes(); await createSchedulesHandlers(deps).listSchedules( @@ -553,7 +554,28 @@ describe('deferred erase retry', () => { // Fire-and-forget, so let the microtask chain settle before asserting. await new Promise((resolve) => setImmediate(resolve)); - expect(deps.methods.eraseScheduleIfDrained).toHaveBeenCalledWith('stranded-1'); + // The service delete (abort + settle + erase), not a bare erase probe: a schedule + // stranded mid-drain with a still-active run needs the abort re-driven too. + expect(deps.deleteSchedule).toHaveBeenCalledWith('stranded-1', 'user-1'); + }); + + it('keeps listing even when a stranded deletion re-drive rejects', async () => { + const deps = makeCreateDeps({ + deleteSchedule: jest.fn(async () => { + throw new Error('still draining'); + }), + } as Partial); + (deps.methods.getSchedulesByUser as jest.Mock) = jest.fn(async () => []); + (deps.methods.getDeletingScheduleIds as jest.Mock) = jest.fn(async () => ['stranded-1']); + const { res, captured } = makeRes(); + + await createSchedulesHandlers(deps).listSchedules( + { user: { id: 'user-1' } } as unknown as ServerRequest, + res, + ); + await new Promise((resolve) => setImmediate(resolve)); + + expect(captured.body).toEqual(expect.objectContaining({ schedules: [] })); }); }); @@ -645,6 +667,85 @@ describe('updateSchedule refuses field-less payloads', () => { }); }); +describe('updateSchedule re-enable attachment revalidation', () => { + const disabledWithFiles = () => + ({ + id: 'sched-1', + enabled: false, + agent_id: 'agent-1', + cadence: { type: 'daily', hour: 9, minute: 0 }, + timezone: 'UTC', + nextRunAt: new Date('2026-07-31T09:00:00Z'), + configRevision: 3, + file_ids: ['file-a', 'file-b'], + }) as unknown as ISchedule; + + const makeReEnableReq = () => + ({ + params: { id: 'sched-1' }, + body: { enabled: true }, + user: { id: 'user-1', tenantId: 't1', role: 'USER' }, + }) as unknown as ServerRequest; + + it('refuses re-enabling when a stored attachment is no longer owned', async () => { + const deps = makeCreateDeps({ + isUserDeleting: jest.fn(async () => false), + filterOwnedFileIds: jest.fn(async () => ['file-a']), + } as Partial); + (deps.methods.getScheduleById as jest.Mock).mockResolvedValue(disabledWithFiles()); + const { res, captured } = makeRes(); + + await createSchedulesHandlers(deps).updateSchedule(makeReEnableReq(), res); + + // The bounded upload hold only renews while the schedule fires, so a long- + // disabled schedule can have lost its uploads; silently firing without them + // is worse than telling the owner to replace the attachments. + expect(captured.status).toBe(400); + expect(deps.methods.updateScheduleById).not.toHaveBeenCalled(); + }); + + it('renews the retention hold on the stored attachments before committing', async () => { + const markFilesUsed = jest.fn(async () => undefined); + const deps = makeCreateDeps({ + isUserDeleting: jest.fn(async () => false), + markFilesUsed, + } as Partial); + (deps.methods.getScheduleById as jest.Mock).mockResolvedValue(disabledWithFiles()); + const { res, captured } = makeRes(); + + await createSchedulesHandlers(deps).updateSchedule(makeReEnableReq(), res); + + expect(markFilesUsed).toHaveBeenCalledWith(['file-a', 'file-b'], 'user-1'); + expect(captured.status ?? 200).toBe(200); + expect(deps.methods.updateScheduleById).toHaveBeenCalled(); + }); + + it('skips the stored-attachment recheck when the edit replaces file_ids', async () => { + const filterOwnedFileIds = jest.fn(async (ids: string[]) => ids); + const deps = makeCreateDeps({ + isUserDeleting: jest.fn(async () => false), + filterOwnedFileIds, + } as Partial); + (deps.methods.getScheduleById as jest.Mock).mockResolvedValue(disabledWithFiles()); + const { res, captured } = makeRes(); + + await createSchedulesHandlers(deps).updateSchedule( + { + params: { id: 'sched-1' }, + body: { enabled: true, file_ids: ['file-new'] }, + user: { id: 'user-1', tenantId: 't1', role: 'USER' }, + } as unknown as ServerRequest, + res, + ); + + // Supplied file_ids are validated by validatePayload; the stored list is + // about to be overwritten, so rechecking it would refuse a valid replacement. + expect(captured.status ?? 200).toBe(200); + expect(filterOwnedFileIds).toHaveBeenCalledTimes(1); + expect(filterOwnedFileIds).toHaveBeenCalledWith(['file-new'], 'user-1'); + }); +}); + describe('late-create compensation with a live manual run', () => { it('treats a draining rollback as compensated (the teardown owns it)', async () => { const deps = makeCreateDeps(); diff --git a/packages/api/src/schedules/handlers.ts b/packages/api/src/schedules/handlers.ts index 63cb718f4e..7e15f331b9 100644 --- a/packages/api/src/schedules/handlers.ts +++ b/packages/api/src/schedules/handlers.ts @@ -279,27 +279,29 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH } /** - * Re-drives the deferred erase of the caller's soft-deleted schedules, off the + * Re-drives the FULL deletion of the caller's soft-deleted schedules, off the * response path. * - * A `deleting` row is erased by whichever actor observes it drained: the delete - * request itself, or the terminal outcome write (erase-on-settle). Both are - * best-effort single attempts, and the reconciler that would otherwise retry does not - * exist in the clustered entrypoint — so one transient failure, or a lease that - * outlived the delete, strands a hidden row holding the user's prompt indefinitely. - * The row is hidden from the list, so the owner cannot even retry it themselves. + * A `deleting` row is settled and erased by the delete request itself, or by the + * terminal outcome write (erase-on-settle). Both are best-effort, and the + * reconciler that would otherwise retry does not exist in the clustered + * entrypoint. Meanwhile the row is HIDDEN from the owner's list, so once a delete + * answers 503-unconfirmed there is no UI/API-list handle left to retry the drain + * with — the "please retry" the response asks for has nothing to click. * - * A read the owner performs anyway is the cheapest place to retry: bounded, scoped to - * their own rows, and a no-op when nothing is deleting (the erase re-checks drained-ness - * itself, so this can never race a live run). + * A read the owner performs anyway is therefore the re-driver: bounded, scoped to + * their own rows, a no-op when nothing is deleting. Re-driving the WHOLE delete + * (abort, settle, erase) rather than only the erase is what un-strands a row whose + * active run never settled; every step is idempotent and evidence-guarded, so + * repeated polls race harmlessly. */ - function retryDeferredErases(userId: string): void { + function retryDeferredDeletions(userId: string): void { void deps.methods .getDeletingScheduleIds(userId, DEFERRED_ERASE_RETRY_LIMIT) .then((ids) => - Promise.all(ids.map((id) => deps.methods.eraseScheduleIfDrained(id).catch(() => false))), + Promise.all(ids.map((id) => deps.deleteSchedule(id, userId).catch(() => 'unconfirmed'))), ) - .catch((err) => logger.warn('[schedules] deferred erase retry failed', err)); + .catch((err) => logger.warn('[schedules] deferred deletion retry failed', err)); } async function listSchedules(req: ServerRequest, res: Response): Promise { @@ -307,7 +309,7 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH deps.methods.getSchedulesByUser(requestUser(req).id), deps.getLimits(requestUser(req)), ]); - retryDeferredErases(requestUser(req).id); + retryDeferredDeletions(requestUser(req).id); res.json({ schedules: schedules.map(toWireSchedule), limits: { maxPerUser: limits.maxPerUser }, @@ -653,6 +655,26 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH res.status(500).json({ error: 'Failed to retain schedule attachments' }); return; } + // Re-enabling with STORED attachments: the bounded upload hold only renews while + // the schedule fires, so a schedule that sat disabled past the hold can have lost + // its uploads. Validate the effective list and renew its hold now, mirroring the + // stored-agent recheck above — otherwise the re-enable succeeds and the next run + // silently fires without the missing files instead of telling the user to + // replace them. + if (reEnabled && parsed.data.file_ids == null && existing.file_ids?.length) { + const stillOwned = await deps.filterOwnedFileIds(existing.file_ids, user.id); + if (stillOwned.length !== existing.file_ids.length) { + res.status(400).json({ + error: + 'One or more attached files are no longer available. Replace the attachments before re-enabling.', + }); + return; + } + if (!(await retainFiles(existing.file_ids, user.id))) { + res.status(500).json({ error: 'Failed to retain schedule attachments' }); + return; + } + } // FENCED on the revision this edit was computed from. `nextRunAt` above is derived // from (cadence, timezone) resolved against the row read at the top of this handler, // so two overlapping edits — one changing cadence, one changing timezone — would diff --git a/packages/data-schemas/src/methods/schedule.methods.spec.ts b/packages/data-schemas/src/methods/schedule.methods.spec.ts index 3deaff9784..17f30c6db5 100644 --- a/packages/data-schemas/src/methods/schedule.methods.spec.ts +++ b/packages/data-schemas/src/methods/schedule.methods.spec.ts @@ -748,6 +748,28 @@ describe('recordRunOutcome idempotency + crash-retry (bookkeeping)', () => { // No longer surfaced as needing bookkeeping. expect(await methods.getUnbookkeptRuns(new Date(Date.now() + 1000), 100)).toHaveLength(0); }); + + it('replayed bookkeeping projects the ORIGINAL firedAt, not the replay time', async () => { + const schedule = await methods.createSchedule(scheduleData()); + const originalFiredAt = new Date('2026-07-20T12:00:03Z'); + await methods.insertScheduleRun(runData(schedule, { scheduledFor, firedAt: originalFiredAt })); + await ScheduleRun.updateOne( + { scheduleId: schedule.id, scheduledFor }, + { $set: { status: 'success', bookkept: false } }, + ); + + // The reconciler replays minutes (or several retries) later — lastRun must + // still show when the run actually fired, not when recovery caught up. + await methods.finalizeBookkeeping({ + scheduleId: schedule.id, + scheduledFor, + status: 'success', + autoDisableAfterFailures: 3, + }); + + const updated = await getSchedule(schedule.id); + expect(updated.lastRun?.firedAt?.toISOString()).toBe(originalFiredAt.toISOString()); + }); }); describe('bookkeeping policy is crash-retryable / streak-correct', () => { diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index f274ce0538..56e6831c01 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -1201,15 +1201,18 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche // re-running bookkeeping, so nothing else would ever repair a half-applied skip. const inserted = await insertScheduleRun({ ...data, firedAt, bookkept: false }); let rowRevision = inserted?.configRevision; + let rowFiredAt = firedAt; if (inserted == null) { const existing = await ScheduleRun() .findOne({ scheduleId: data.scheduleId, scheduledFor: data.scheduledFor }) - .select('status configRevision') - .lean>(); + .select('status configRevision firedAt') + .lean>(); if (existing == null || existing.status !== data.status) { return; } rowRevision = existing.configRevision; + // A retry re-projects the ORIGINAL occurrence, not the retry moment. + rowFiredAt = existing.firedAt ?? firedAt; } // Same single seam as recordRunOutcome: the config fence is DERIVED from the row, // never passed by the caller. A skip decided under an older owner config must not @@ -1219,7 +1222,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche await applyBalanceSkipBookkeeping({ scheduleId: data.scheduleId, scheduledFor: data.scheduledFor, - firedAt, + firedAt: rowFiredAt, rowRevision, balanceSkipDisableThreshold, }); @@ -1227,7 +1230,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche await applyOverlapSkipBookkeeping({ scheduleId: data.scheduleId, scheduledFor: data.scheduledFor, - firedAt, + firedAt: rowFiredAt, rowRevision, }); } @@ -1319,15 +1322,19 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche async function finalizeBookkeeping(params: RecordRunOutcomeParams): Promise { // Same single seam as recordRunOutcome: derive the config fence from the row, so the // crash-retry path cannot apply bookkeeping the inline path would have refused. + // The row's ORIGINAL firedAt is reused too: this replay runs minutes (or many + // retries) after the fact, and stamping the replay time onto lastRun would keep + // walking the card's timestamp forward every time the recovery path fires. const run = await ScheduleRun() .findOne({ scheduleId: params.scheduleId, scheduledFor: params.scheduledFor }) - .select('configRevision') - .lean>(); + .select('configRevision firedAt') + .lean>(); + const firedAt = run?.firedAt ?? new Date(); if (params.status === 'skipped_balance') { await applyBalanceSkipBookkeeping({ scheduleId: params.scheduleId, scheduledFor: params.scheduledFor, - firedAt: new Date(), + firedAt, conversationId: params.conversationId, rowRevision: run?.configRevision, balanceSkipDisableThreshold: params.balanceSkipDisableThreshold, @@ -1336,13 +1343,13 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche await applyOverlapSkipBookkeeping({ scheduleId: params.scheduleId, scheduledFor: params.scheduledFor, - firedAt: new Date(), + firedAt, rowRevision: run?.configRevision, }); } else { await applyTerminalBookkeeping({ ...params, - firedAt: new Date(), + firedAt, expectConfigRevision: run?.configRevision, }); }