From 74f7af8923a2cf4120d8293b489cd0fce6ebad5f Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 20:33:14 +0000 Subject: [PATCH] fix(schedules): require definite delivery failure, extract its message, converge deferred Stops MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Third Codex round. Three findings in code from this closeout, plus one pre-existing P1 that is a one-line operator-safety fix. P1 — dead-delivery settlement demanded too little evidence. `dead` does not prove a request was rejected: the trigger host marks response timeouts and invalid success responses `certainty: 'ambiguous'`, and the engine dead-letters those once retries are exhausted. The erasure sweep treated every dead letter as positive evidence, so an ambiguous one sitting over a generation a peer had accepted could terminalize the run and release its capacity mid-flight. It now settles only on a DEFINITE rejection, unless job absence is deployment-authoritative (safe topology), where the confirmed-absent job is itself the evidence. P1 — `lastError` is an `AgentTriggerDeliveryFailure` object, not a string. A duplicated local interface declared it `string` (against CLAUDE.md's no-duplicate- types rule), so both the sweep and the engine reconciler passed the object into the String-typed run/schedule `error` fields; Mongoose would reject the cast, the per-row catch would swallow it, and the run would keep its global capacity slot. The dep type now reuses the canonical `AgentTriggerDeliveryFailure` and both call sites pass `.message`. Re-typing immediately surfaced a stale test that had asserted a string. P1 (pre-existing) — the base-config global stop is honored in `getLimits` via `isRuntimeDisabled` rather than a literal `=== false`. The stop has two shapes, and deepMerge turns base `{ use: false }` plus a principal override of `true` into `{ use: true }`, so the literal check reported the feature enabled and Run Now dispatched straight through fireSchedule, bypassing the operator's emergency stop. Now the same predicate the engine gate already uses. P2 — a Stop whose settlement DEFERRED past the poll budget had no convergence path where no reconciler is armed. `acknowledgeScheduledStopPersistence` now optionally re-drives the terminal outcome once the barrier clears; `recordRunOutcome` is match-guarded and idempotent, so an owner that already settled makes it a no-op. The abort route passes it for a running generation; a paused job still settles explicitly. Tests: ambiguous dead letters refused under unsafe topology but settled when absence is authoritative, definite rejections settled either way, the failure message carried through, and the abort route's re-drive present for running / absent for paused. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq --- .../routes/agents/__tests__/abort.spec.js | 38 +++++++- api/server/routes/agents/index.js | 13 +++ packages/api/src/schedules/engine.spec.ts | 11 ++- packages/api/src/schedules/engine.ts | 2 +- packages/api/src/schedules/erasure.spec.ts | 94 ++++++++++++++++++- packages/api/src/schedules/erasure.ts | 15 ++- packages/api/src/schedules/service.ts | 30 +++++- packages/api/src/schedules/types.ts | 8 +- 8 files changed, 195 insertions(+), 16 deletions(-) diff --git a/api/server/routes/agents/__tests__/abort.spec.js b/api/server/routes/agents/__tests__/abort.spec.js index d05334fcb1..43516fbb16 100644 --- a/api/server/routes/agents/__tests__/abort.spec.js +++ b/api/server/routes/agents/__tests__/abort.spec.js @@ -900,10 +900,16 @@ describe('Agent Abort Endpoint', () => { expect(mockBeginScheduledStop.mock.invocationCallOrder[0]).toBeLessThan( mockGenerationJobManager.abortJob.mock.invocationCallOrder[0], ); - expect(mockAcknowledgeScheduledStopPersistence).toHaveBeenCalledWith({ - scheduleId: 's1', - scheduledFor: '2026-01-01T00:00:00.000Z', - }); + // The ack also carries a terminal outcome to re-drive: the owner calls + // recordScheduleOutcome once, and if that call's Stop barrier deferred (slow + // beforePublish), nothing would settle the run where no reconciler is armed. + expect(mockAcknowledgeScheduledStopPersistence).toHaveBeenCalledWith( + expect.objectContaining({ + scheduleId: 's1', + scheduledFor: '2026-01-01T00:00:00.000Z', + settle: expect.objectContaining({ status: 'interrupted' }), + }), + ); expect(mockGenerationJobManager.abortJob.mock.invocationCallOrder[0]).toBeLessThan( mockAcknowledgeScheduledStopPersistence.mock.invocationCallOrder[0], ); @@ -945,6 +951,30 @@ describe('Agent Abort Endpoint', () => { expect(mockAcknowledgeScheduledStopPersistence).not.toHaveBeenCalled(); }); + it('does not re-drive settlement for a PAUSED job, which settles explicitly', async () => { + mockGenerationJobManager.getJob.mockResolvedValue({ + ...scheduledJob, + status: 'requires_action', + }); + mockGenerationJobManager.abortJob.mockResolvedValue({ + success: true, + jobData: { conversationId: 'conv-1' }, + content: [], + text: '', + }); + + const response = await request(app) + .post('/api/agents/chat/abort') + .set('X-LibreChat-Generation-Protocol', '2') + .send({ conversationId: 'conv-1', generationProtocolVersion: 2 }); + + expect(response.status).toBe(200); + const ack = mockAcknowledgeScheduledStopPersistence.mock.calls[0][0]; + expect(ack.settle).toBeUndefined(); + // The paused path records its own outcome explicitly. + expect(mockRecordScheduleOutcome).toHaveBeenCalled(); + }); + it('leaves non-scheduled aborts untouched by the Stop protocol', async () => { mockGenerationJobManager.getJob.mockResolvedValue({ status: 'running', diff --git a/api/server/routes/agents/index.js b/api/server/routes/agents/index.js index baf380c667..1d73ea5d98 100644 --- a/api/server/routes/agents/index.js +++ b/api/server/routes/agents/index.js @@ -908,6 +908,19 @@ router.post('/chat/abort', configMiddleware, async (req, res, next) => { await acknowledgeScheduledStopPersistence({ scheduleId: stopScheduleId, scheduledFor: stopScheduledFor, + // Re-drive the terminal outcome from here for a RUNNING generation: its owner + // calls recordScheduleOutcome once, and if that call's Stop barrier deferred + // (slow beforePublish), nothing would settle the run where no schedule + // reconciler is armed. recordRunOutcome is match-guarded and idempotent, so an + // owner that already settled makes this a no-op. A paused job is settled + // explicitly below and needs no re-drive here. + ...(job.status !== 'requires_action' && { + settle: { + status: 'interrupted', + conversationId: job.metadata?.conversationId ?? jobStreamId, + error: 'Scheduled run was stopped', + }, + }), }); } diff --git a/packages/api/src/schedules/engine.spec.ts b/packages/api/src/schedules/engine.spec.ts index 3865bffbf1..373c00bc41 100644 --- a/packages/api/src/schedules/engine.spec.ts +++ b/packages/api/src/schedules/engine.spec.ts @@ -254,7 +254,16 @@ describe('reconciliation consults the durable trigger delivery', () => { it('settles a DEAD delivery as error promptly, before the 30-minute orphan age', async () => { const methods = await runReconcile(joblessRun(YOUNG()), async () => ({ status: 'dead', - lastError: 'rate limited', + // The durable record's lastError is an AgentTriggerDeliveryFailure OBJECT, not a + // string — the recorded outcome must carry its `message`, or the String-typed + // Mongoose field rejects the cast and the run keeps its capacity slot. + lastError: { + code: 'rate_limited', + message: 'rate limited', + certainty: 'definite' as const, + retryable: false, + attemptedAt: new Date(), + }, })); expect(methods.recordRunOutcome).toHaveBeenCalledWith( expect.objectContaining({ scheduleId: 's1', status: 'error', error: 'rate limited' }), diff --git a/packages/api/src/schedules/engine.ts b/packages/api/src/schedules/engine.ts index 4a7680b895..976e5ee0a4 100644 --- a/packages/api/src/schedules/engine.ts +++ b/packages/api/src/schedules/engine.ts @@ -262,7 +262,7 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { // ordinary outcome path — no need to wait the 30-minute orphan age. await finalize( 'error', - delivery?.lastError ?? 'Scheduled delivery failed before running', + delivery?.lastError?.message ?? 'Scheduled delivery failed before running', ); continue; } diff --git a/packages/api/src/schedules/erasure.spec.ts b/packages/api/src/schedules/erasure.spec.ts index 4d10a9a099..cc50fce2a4 100644 --- a/packages/api/src/schedules/erasure.spec.ts +++ b/packages/api/src/schedules/erasure.spec.ts @@ -82,6 +82,12 @@ describe('schedule erasure fallback owner-death evidence', () => { }); }); +const DEFINITE = { + code: 'rejected', + message: 'rejected before running', + certainty: 'definite' as const, +}; + describe('topology-safe dead-delivery convergence', () => { beforeEach(() => { jest.useFakeTimers(); @@ -95,7 +101,10 @@ describe('topology-safe dead-delivery convergence', () => { async function convergeOnce(options: { run?: Partial; - delivery?: { status: string; lastError?: string } | null; + delivery?: { + status: string; + lastError?: { code: string; message: string; certainty: 'definite' | 'ambiguous' }; + } | null; job?: JobState | null; canInferOwnerDeathFromMissingJob?: boolean; }) { @@ -124,7 +133,10 @@ describe('topology-safe dead-delivery convergence', () => { it('settles a dead delivery as error even in an unsafe topology (positive evidence)', async () => { const { methods } = await convergeOnce({ - delivery: { status: 'dead', lastError: 'blocked by moderation' }, + delivery: { + status: 'dead', + lastError: { code: 'blocked', message: 'blocked by moderation', certainty: 'definite' }, + }, }); expect(methods.recordRunOutcome).toHaveBeenCalledWith( expect.objectContaining({ @@ -142,7 +154,7 @@ describe('topology-safe dead-delivery convergence', () => { it('never settles while an identity-matched generation still owns the reservation', async () => { const { methods, getTriggerDelivery } = await convergeOnce({ - delivery: { status: 'dead' }, + delivery: { status: 'dead', lastError: DEFINITE }, job: { status: 'running', scheduleId: 'schedule-1', @@ -156,7 +168,7 @@ describe('topology-safe dead-delivery convergence', () => { it('defers a run whose abort is still in flight', async () => { const { methods } = await convergeOnce({ run: { abortRequestedAt: new Date() } as Partial, - delivery: { status: 'dead' }, + delivery: { status: 'dead', lastError: DEFINITE }, }); expect(methods.recordRunOutcome).not.toHaveBeenCalled(); }); @@ -164,9 +176,81 @@ describe('topology-safe dead-delivery convergence', () => { it('ignores a legacy reservation carrying no deliveryKey', async () => { const { methods, getTriggerDelivery } = await convergeOnce({ run: { deliveryKey: undefined } as Partial, - delivery: { status: 'dead' }, + delivery: { status: 'dead', lastError: DEFINITE }, }); expect(getTriggerDelivery).not.toHaveBeenCalled(); expect(methods.recordRunOutcome).not.toHaveBeenCalled(); }); }); + +describe('dead-delivery certainty fence', () => { + beforeEach(() => { + jest.useFakeTimers(); + jest.spyOn(Math, 'random').mockReturnValue(0); + }); + + afterEach(() => { + jest.useRealTimers(); + jest.restoreAllMocks(); + }); + + async function sweepWithDelivery(options: { + certainty: 'definite' | 'ambiguous'; + canInferOwnerDeathFromMissingJob: boolean; + }) { + const methods = { + getDeletingSchedules: jest.fn(async () => []), + getActiveRunsForSchedule: jest.fn(async () => []), + recordRunOutcome: jest.fn(async () => undefined), + eraseScheduleIfDrained: jest.fn(async () => false), + markEraseAttempted: jest.fn(async () => undefined), + getRunsForReconciliation: jest.fn(async () => [oldRun({ deliveryKey: 'dk-1' } as never)]), + }; + const sweep = startScheduleErasureSweep({ + methods: methods as unknown as ScheduleMethods, + getJobStatus: jest.fn(async () => null), + getTriggerDelivery: (async () => ({ + status: 'dead', + lastError: { code: 'x', message: 'timed out', certainty: options.certainty }, + })) as never, + canInferOwnerDeathFromMissingJob: options.canInferOwnerDeathFromMissingJob, + }); + await jest.advanceTimersByTimeAsync(5 * 60_000); + sweep.stop(); + return methods; + } + + /** + * `dead` is not proof of rejection: the trigger host marks response timeouts and invalid + * success responses `ambiguous`, and the engine dead-letters those after exhausting + * retries — so an ambiguous dead letter can sit over a generation a peer accepted and is + * still running. Settling it here would release capacity mid-run. + */ + it('refuses an AMBIGUOUS dead letter when absence is not deployment-authoritative', async () => { + const methods = await sweepWithDelivery({ + certainty: 'ambiguous', + canInferOwnerDeathFromMissingJob: false, + }); + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + }); + + it('settles an ambiguous dead letter once job absence IS deployment-authoritative', async () => { + const methods = await sweepWithDelivery({ + certainty: 'ambiguous', + canInferOwnerDeathFromMissingJob: true, + }); + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ status: 'error', error: 'timed out' }), + ); + }); + + it('settles a DEFINITE rejection even in an unsafe topology', async () => { + const methods = await sweepWithDelivery({ + certainty: 'definite', + canInferOwnerDeathFromMissingJob: false, + }); + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ status: 'error', error: 'timed out' }), + ); + }); +}); diff --git a/packages/api/src/schedules/erasure.ts b/packages/api/src/schedules/erasure.ts index 01b869a0cf..902dbfde5b 100644 --- a/packages/api/src/schedules/erasure.ts +++ b/packages/api/src/schedules/erasure.ts @@ -170,12 +170,25 @@ async function settleDeadDeliveries(deps: ScheduleErasureDeps): Promise { if (delivery?.status !== 'dead') { continue; } + // `dead` alone is NOT proof the request was rejected. The trigger host marks + // response timeouts and invalid success responses `ambiguous`, and the engine + // dead-letters those once retries are exhausted — so an ambiguous dead letter can + // sit over a generation a peer accepted and is still running. Settle only on a + // DEFINITE rejection, unless this process can observe generation absence + // deployment-wide (safe topology), where the confirmed-absent job above is itself + // authoritative evidence. + if (delivery.lastError?.certainty !== 'definite' && !deps.canInferOwnerDeathFromMissingJob) { + continue; + } await deps.methods.recordRunOutcome({ scheduleId: run.scheduleId, scheduledFor: run.scheduledFor, status: 'error', conversationId: run.conversationId, - error: delivery.lastError ?? 'Scheduled delivery failed before running', + // `lastError` is an AgentTriggerDeliveryFailure, not a string: passing the object + // into the String-typed run/schedule error fields fails the Mongoose cast, and the + // per-row catch would swallow it while the run kept its capacity slot. + error: delivery.lastError?.message ?? 'Scheduled delivery failed before running', autoDisableAfterFailures: Number.MAX_SAFE_INTEGER, }); } catch (err) { diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index 6f4887d416..163562751d 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -196,6 +196,9 @@ export interface SchedulesService { acknowledgeScheduledStopPersistence: (input: { scheduleId: string; scheduledFor: string | Date; + /** Optional terminal outcome to re-drive once the barrier clears, so a settlement the + * owner deferred past its poll budget still converges where no reconciler is armed. */ + settle?: { status: ScheduleRunOutcomeStatus; conversationId?: string; error?: string }; }) => Promise; /** Re-enters a paused occurrence into the DB-enforced global/same-schedule * capacity set before its approval job is allowed to resume. */ @@ -312,9 +315,15 @@ export function createSchedulesService( // principal override. Without this a tenant/role/user override resolving to // enabled would let the sidebar and CRUD handlers admit schedules that // isGloballyDisabled() correctly refuses to ever fire. + // isRuntimeDisabled, NOT `=== false`: the base stop has TWO shapes (`false` and + // `{ use: false }`), and deepMerge turns a base `{ use: false, ... }` plus a principal + // override of `true` into `{ use: true, ... }`. A literal-false check missed that, so + // getLimits reported the feature enabled and Run Now dispatched straight through + // fireSchedule — bypassing the operator's object-form emergency stop. Same predicate + // the engine gate (isGloballyDisabled) already uses. if ( user != null && - (await deps.getAppConfig({ baseOnly: true }))?.interfaceConfig?.schedules === false + isRuntimeDisabled((await deps.getAppConfig({ baseOnly: true }))?.interfaceConfig?.schedules) ) { return { ...DEFAULT_SCHEDULE_LIMITS, enabled: false }; } @@ -929,14 +938,33 @@ export function createSchedulesService( async function acknowledgeScheduledStopPersistence({ scheduleId, scheduledFor, + settle, }: { scheduleId: string; scheduledFor: string | Date; + /** Terminal outcome to (re-)drive once the barrier clears. Supplied by the Stop route + * so a settlement its owner already DEFERRED past the poll budget still converges. */ + settle?: { status: ScheduleRunOutcomeStatus; conversationId?: string; error?: string }; }): Promise { if (!scheduleId || !scheduledFor) { return; } await methods.markRunAbortPersisted(scheduleId, new Date(scheduledFor)); + if (settle == null) { + return; + } + // The owner calls recordScheduleOutcome ONCE. If its Stop barrier deferred (slow + // beforePublish), nothing would re-drive it where no schedule reconciler is armed — + // the run would stay `started`, its job preserved, and its global capacity slot held. + // Now that the barrier is acknowledged, settle from here; recordRunOutcome is + // match-guarded and idempotent, so a run the owner already settled is a no-op. + await recordScheduleOutcome({ + scheduleId, + scheduledFor, + status: settle.status, + conversationId: settle.conversationId, + error: settle.error, + }).catch((err) => logger.warn('[schedules] post-acknowledgement settlement failed:', err)); } async function isScheduleLive( diff --git a/packages/api/src/schedules/types.ts b/packages/api/src/schedules/types.ts index b3aa088f66..4a1c250751 100644 --- a/packages/api/src/schedules/types.ts +++ b/packages/api/src/schedules/types.ts @@ -2,6 +2,7 @@ import type { ScheduleMethods, ISchedule, AgentTriggerDeliveryStatus, + AgentTriggerDeliveryFailure, } from '@librechat/data-schemas'; import type { Types } from 'mongoose'; import type { AgentTriggerEnqueueOptions, AgentTriggerEnvelope } from '../agents/triggers'; @@ -147,9 +148,10 @@ export interface ScheduleEngineDeps { * `Retry-After` can defer up to 24h) or a dead-letter (`dead`, with its `lastError`) * apart from a genuinely orphaned jobless run. Null when no delivery record exists. */ - getTriggerDelivery: ( - deliveryKey: string, - ) => Promise<{ status: AgentTriggerDeliveryStatus; lastError?: string } | null>; + getTriggerDelivery: (deliveryKey: string) => Promise<{ + status: AgentTriggerDeliveryStatus; + lastError?: AgentTriggerDeliveryFailure; + } | null>; /** Runs fn inside the owner's tenant ALS context. */ runInTenantContext: (user: ScheduleUserContext, fn: () => Promise) => Promise; /**