mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-22 10:02:41 +00:00
fix(schedules): require definite delivery failure, extract its message, converge deferred Stops
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq
This commit is contained in:
parent
ef4192a408
commit
74f7af8923
8 changed files with 195 additions and 16 deletions
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
},
|
||||
}),
|
||||
});
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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' }),
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<IScheduleRun>;
|
||||
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<IScheduleRun>,
|
||||
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<IScheduleRun>,
|
||||
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' }),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -170,12 +170,25 @@ async function settleDeadDeliveries(deps: ScheduleErasureDeps): Promise<void> {
|
|||
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) {
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
/** 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<void> {
|
||||
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(
|
||||
|
|
|
|||
|
|
@ -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: <T>(user: ScheduleUserContext, fn: () => Promise<T>) => Promise<T>;
|
||||
/**
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue