diff --git a/api/server/controllers/agents/request.js b/api/server/controllers/agents/request.js index f1dde7c8ef..800643868d 100644 --- a/api/server/controllers/agents/request.js +++ b/api/server/controllers/agents/request.js @@ -977,14 +977,6 @@ 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) { - // 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; @@ -996,6 +988,22 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit } delete response.databasePromise; } + // UNBLOCK the title BEFORE the persistence barrier below: `addTitle` waits + // on `convoReady` before persisting, so awaiting `immediateTitlePromise` + // with `convoReady` still pending deadlocks this branch. The conversation + // row was saved just above, so a title that already finished generating may + // persist now; the abort only cancels a still-in-flight title model call. + titleAbortController.abort(); + acceptsTitleEvents = false; + resolveConvoReady(); + // 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(); // BaseClient saved the response as completed (unfinished:false), but the turn // is paused awaiting a decision. Re-mark it unfinished so an expired / never- // resumed approval doesn't leave a "finished" response in history; the resume @@ -1053,9 +1061,6 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit } } } - titleAbortController.abort(); - acceptsTitleEvents = false; - resolveConvoReady(); // handleRunInterrupt already released the concurrency slot the moment it paused // (so a fast /resume isn't 429'd); only release here if that didn't happen. // Always run the MCP request-context cleanup. diff --git a/api/server/routes/agents/index.js b/api/server/routes/agents/index.js index b324257607..8b426c1558 100644 --- a/api/server/routes/agents/index.js +++ b/api/server/routes/agents/index.js @@ -473,6 +473,11 @@ router.post('/chat/abort', configMiddleware, async (req, res) => { liveJob = await GenerationJobManager.getJob(jobStreamId); } catch (err) { logger.error(`[AgentStream] Could not verify abort state: ${jobStreamId}`, err); + // This attempt is over either way — nothing was stopped, so this route has no + // persistence left to perform. Leaving the stamp unresolved would make every + // retry answer 'in_progress' (a false success) and hold the owner's settlement + // barrier + the reconciler fence for the full stale window. + await resolveStopAttempt(); return res .status(503) .json({ error: 'Could not verify the generation state. Please retry.', aborted: null }); diff --git a/packages/api/src/schedules/handlers.spec.ts b/packages/api/src/schedules/handlers.spec.ts index 314304e72c..9f4d0d284d 100644 --- a/packages/api/src/schedules/handlers.spec.ts +++ b/packages/api/src/schedules/handlers.spec.ts @@ -545,6 +545,7 @@ describe('deferred deletion retry', () => { } as Partial); (deps.methods.getSchedulesByUser as jest.Mock) = jest.fn(async () => []); (deps.methods.getDeletingScheduleIds as jest.Mock) = jest.fn(async () => ['stranded-1']); + (deps.methods.markEraseAttempted as jest.Mock) = jest.fn(async () => undefined); const { res } = makeRes(); await createSchedulesHandlers(deps).listSchedules( @@ -557,6 +558,8 @@ describe('deferred deletion retry', () => { // 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'); + // Stamped attempted so the bounded window rotates past rows that stay unconfirmed. + expect(deps.methods.markEraseAttempted).toHaveBeenCalledWith(['stranded-1']); }); it('keeps listing even when a stranded deletion re-drive rejects', async () => { @@ -567,6 +570,7 @@ describe('deferred deletion retry', () => { } as Partial); (deps.methods.getSchedulesByUser as jest.Mock) = jest.fn(async () => []); (deps.methods.getDeletingScheduleIds as jest.Mock) = jest.fn(async () => ['stranded-1']); + (deps.methods.markEraseAttempted as jest.Mock) = jest.fn(async () => undefined); const { res, captured } = makeRes(); await createSchedulesHandlers(deps).listSchedules( diff --git a/packages/api/src/schedules/handlers.ts b/packages/api/src/schedules/handlers.ts index 7e15f331b9..0538c82d1b 100644 --- a/packages/api/src/schedules/handlers.ts +++ b/packages/api/src/schedules/handlers.ts @@ -298,9 +298,19 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH function retryDeferredDeletions(userId: string): void { void deps.methods .getDeletingScheduleIds(userId, DEFERRED_ERASE_RETRY_LIMIT) - .then((ids) => - Promise.all(ids.map((id) => deps.deleteSchedule(id, userId).catch(() => 'unconfirmed'))), - ) + .then(async (ids) => { + if (ids.length === 0) { + return; + } + // Stamp BEFORE attempting: the read window is least-recently-attempted + // first, so rows this pass touches rotate to the back and an owner with + // more stuck rows than the limit reaches all of them across successive + // lists instead of re-driving the same unconfirmable few forever. + await deps.methods.markEraseAttempted(ids); + await Promise.all( + ids.map((id) => deps.deleteSchedule(id, userId).catch(() => 'unconfirmed')), + ); + }) .catch((err) => logger.warn('[schedules] deferred deletion retry failed', err)); } diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index 36d401cca4..026f14a537 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -11,6 +11,7 @@ import type { TMessageContentParts, TContextUsageEvent, TTokenUsageEvent, + TPendingSteer, Agents, } from 'librechat-data-provider'; import type { StandardGraph } from '@librechat/agents'; @@ -1619,20 +1620,34 @@ class GenerationJobManagerClass { * the createdAt guard keeps it off a replacement job's queue. Runs only * after WINNING the terminal CAS: a losing abort must not close or drain * the winner's queue. */ - const pendingSteers = ( - await this.jobStore.closeAndDrainSteers(streamId, jobData.createdAt) - ).map(toPendingSteer); - // No-subscriber recovery: the abort response/final are transient, so park - // the leftovers for /chat/status claim-on-read within the recovery TTL. - await this.steering.park( - streamId, - pendingSteers, - { - userId: jobData.userId, - tenantId: jobData.tenantId, - }, - jobData.createdAt, - ); + // BEST-EFFORT from here to the abort signal: the terminal CAS above is already + // durable, so a rejection in drain/park would exit before `emitAbort` and the + // local `abortController.abort()` — leaving a job every retry sees as terminal + // while the generation keeps running and billing. Losing the queue report + // (steers re-surface via /chat/status or expire by TTL) is strictly cheaper + // than losing the stop signal. + let pendingSteers: TPendingSteer[] = []; + try { + pendingSteers = (await this.jobStore.closeAndDrainSteers(streamId, jobData.createdAt)).map( + toPendingSteer, + ); + // No-subscriber recovery: the abort response/final are transient, so park + // the leftovers for /chat/status claim-on-read within the recovery TTL. + await this.steering.park( + streamId, + pendingSteers, + { + userId: jobData.userId, + tenantId: jobData.tenantId, + }, + jobData.createdAt, + ); + } catch (err) { + logger.error( + `[GenerationJobManager] Steer drain/park failed during abort for ${streamId}; continuing to signal:`, + err, + ); + } /** Final event for abort */ const userMessageId = jobData.userMessage?.messageId; diff --git a/packages/api/src/stream/__tests__/steering.spec.ts b/packages/api/src/stream/__tests__/steering.spec.ts index f2a7a8daa0..951c4d5eb9 100644 --- a/packages/api/src/stream/__tests__/steering.spec.ts +++ b/packages/api/src/stream/__tests__/steering.spec.ts @@ -464,6 +464,25 @@ describe('SteeringLifecycle via GenerationJobManager.steering (in-memory)', () = expect(await manager.steering.peek(streamId)).toEqual([]); }); + test('abortJob still signals and finalizes when the steer drain fails', async () => { + // The terminal CAS is already durable when the drain runs; a drain rejection + // must not exit before the abort signal, or every retry sees a terminal job + // while the generation keeps running and billing. + const streamId = 'steer-abort-drain-fails'; + await manager.createJob(streamId, 'user-1'); + await manager.steering.enqueue(streamId, buildSteer('stranded but non-fatal')); + jest + .spyOn(jobStore, 'closeAndDrainSteers') + .mockRejectedValueOnce(new Error('transient store failure')); + + const result = await manager.abortJob(streamId); + + expect(result.success).toBe(true); + expect(result.finalEvent).toMatchObject({ aborted: true }); + expect(result.pendingSteers ?? []).toEqual([]); + await expect(jobStore.getJob(streamId)).resolves.toMatchObject({ status: 'aborted' }); + }); + test('abortJob publishes nothing when natural completion wins its terminal CAS', async () => { const streamId = 'steer-abort-loses-terminal-race'; const eventTransport = new InMemoryEventTransport(); diff --git a/packages/data-schemas/src/methods/schedule.methods.spec.ts b/packages/data-schemas/src/methods/schedule.methods.spec.ts index 17f30c6db5..a51edcb36b 100644 --- a/packages/data-schemas/src/methods/schedule.methods.spec.ts +++ b/packages/data-schemas/src/methods/schedule.methods.spec.ts @@ -2138,6 +2138,22 @@ describe('erasure sweep rotation and idempotency-key lookup', () => { expect(rotated[0].id).not.toBe(window[0].id); }); + it('rotates the per-owner deletion-retry window the same way', async () => { + const user = new mongoose.Types.ObjectId(); + await Schedule.create(scheduleData({ user, deleting: true, nextRunAt: undefined })); + await Schedule.create(scheduleData({ user, deleting: true, nextRunAt: undefined })); + + const window = await methods.getDeletingScheduleIds(user, 1); + expect(window).toHaveLength(1); + await methods.markEraseAttempted(window); + + // The list-path retry is bounded too; without least-recently-attempted + // ordering an owner with more stuck rows than the limit re-drives the same + // unconfirmable few on every list and never reaches the rest. + const rotated = await methods.getDeletingScheduleIds(user, 1); + expect(rotated[0]).not.toBe(window[0]); + }); + it('resolves an idempotency key even while its row is draining', async () => { const user = new mongoose.Types.ObjectId(); const schedule = scheduleData({ user, deleting: true, clientRequestId: 'intent-9' }); diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index 56e6831c01..f4a51819c5 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -1527,8 +1527,14 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche userId: string | Types.ObjectId, limit: number, ): Promise { + // Least-recently-attempted first (missing sorts before any date), for the same + // reason the erasure sweep rotates: an unsorted `.limit()` window pins the same + // stuck rows forever once the owner has more deleting rows than the limit, and + // the ones beyond it never get their deletion re-driven. Callers stamp + // markEraseAttempted after each attempt to rotate the window. const rows = await Schedule() .find({ user: userId, deleting: true }) + .sort({ eraseAttemptedAt: 1 }) .select('id') .limit(limit) .lean>>();