diff --git a/api/server/controllers/agents/__tests__/resume.spec.js b/api/server/controllers/agents/__tests__/resume.spec.js index 034c53189e..c38fa9cc9d 100644 --- a/api/server/controllers/agents/__tests__/resume.spec.js +++ b/api/server/controllers/agents/__tests__/resume.spec.js @@ -392,6 +392,82 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => { expect(mockGenerationJobManager.approvals.resolve).not.toHaveBeenCalled(); }); + // `abortJob` reports `success: false` with a REASON on every failure path. Gating on + // the absence of a reason treated an unreached job and a replacement generation as + // confirmed stops, settling the occurrence and pruning a checkpoint on neither. + it.each([ + ['the job vanished before the abort landed', 'job_not_found'], + ['a replacement generation owns the conversation', 'generation_replaced'], + ['the generation is still live', 'job_still_active'], + ])('refuses to settle or prune when %s', async (_label, failureReason) => { + mockGenerationJobManager.getJob.mockResolvedValue(makeScheduledJob()); + mockIsScheduleLive.mockResolvedValue(false); + mockGenerationJobManager.abortJob.mockResolvedValue({ success: false, failureReason }); + + const res = await post(approveBody()); + + expect(res.status).toBe(503); + expect(res.headers['retry-after']).toBe('1'); + expect(res.body).toMatchObject({ code: 'SCHEDULE_STOP_UNCONFIRMED' }); + expect(mockRecordScheduleOutcome).not.toHaveBeenCalled(); + expect(mockDeleteAgentCheckpoint).not.toHaveBeenCalled(); + expect(mockGenerationJobManager.approvals.resolve).not.toHaveBeenCalled(); + }); + + // The exact regression: an abort that reported `success: false` and nothing else was + // read as a confirmed stop, so the occurrence was settled and its checkpoint pruned. + it('refuses to settle or prune on a bare unsuccessful abort with no reason', async () => { + mockGenerationJobManager.getJob.mockResolvedValue(makeScheduledJob()); + mockIsScheduleLive.mockResolvedValue(false); + mockGenerationJobManager.abortJob.mockResolvedValue({ success: false }); + + const res = await post(approveBody()); + + expect(res.status).toBe(503); + expect(res.body).toMatchObject({ code: 'SCHEDULE_STOP_UNCONFIRMED' }); + expect(mockRecordScheduleOutcome).not.toHaveBeenCalled(); + expect(mockDeleteAgentCheckpoint).not.toHaveBeenCalled(); + }); + + it('settles an occurrence whose generation was already terminal and drained', async () => { + mockGenerationJobManager.getJob.mockResolvedValue(makeScheduledJob()); + mockIsScheduleLive.mockResolvedValue(false); + // No transition was needed, but `awaitProviderDrain` still proved the provider + // segment can no longer persist — a stop, just not one this call made. Refusing + // here would 503 a permanently terminal generation on every retry. + mockGenerationJobManager.abortJob.mockResolvedValue({ + success: false, + failureReason: 'already_settled', + }); + + const res = await post(approveBody()); + + expect(res.status).toBe(409); + expect(res.body).toMatchObject({ code: 'SCHEDULE_NO_LONGER_ACTIVE' }); + expect(mockRecordScheduleOutcome).toHaveBeenCalledWith( + expect.objectContaining({ scheduleId: 'schedule-1', status: 'interrupted' }), + ); + expect(mockDeleteAgentCheckpoint).toHaveBeenCalled(); + }); + + it('refuses to settle the stale resume handoff on an unconfirmed stop', async () => { + mockGenerationJobManager.getJob.mockResolvedValue(makeScheduledJob()); + mockFinalizeScheduleResumeClaim.mockResolvedValue(false); + mockGenerationJobManager.abortJob.mockResolvedValue({ + success: false, + failureReason: 'generation_replaced', + }); + + const res = await post(approveBody()); + + expect(res.status).toBe(503); + expect(res.headers['retry-after']).toBe('1'); + expect(res.body).toMatchObject({ code: 'SCHEDULE_STOP_UNCONFIRMED' }); + expect(mockRecordScheduleOutcome).not.toHaveBeenCalled(); + expect(mockDeleteAgentCheckpoint).not.toHaveBeenCalled(); + expect(mockInitializeClient).not.toHaveBeenCalled(); + }); + it('records success after resumed persistence and before terminal publication', async () => { mockGenerationJobManager.getJob.mockResolvedValue(makeScheduledJob()); diff --git a/api/server/controllers/agents/resume.js b/api/server/controllers/agents/resume.js index e4eb76e295..ea14afcb0e 100644 --- a/api/server/controllers/agents/resume.js +++ b/api/server/controllers/agents/resume.js @@ -22,6 +22,7 @@ const { decrementPendingRequest, checkAndIncrementPendingRequest, isSteerPreemptSupported, + isStopConfirmed, toPendingSteer, } = require('@librechat/api'); const { disposeClient } = require('~/server/cleanup'); @@ -608,7 +609,12 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) expectedCreatedAt: job.createdAt, awaitProviderDrain: true, }); - stopped = abortResult != null && abortResult.failureReason == null; + // `success` is the authoritative signal, exactly as the abort route gates. A + // `success: false` result WITHOUT a failure reason no longer exists — an + // unreached job, a replacement, or a lost CAS all report one — so the old + // `failureReason == null` test settled the occurrence and pruned the + // checkpoint on aborts that were never confirmed. + stopped = isStopConfirmed(abortResult); } catch (error) { logger.warn('[ResumeAgentController] Failed to stop inactive scheduled run', error); } @@ -979,7 +985,9 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) expectedCreatedAt: job.createdAt, awaitProviderDrain: true, }); - stopped = abortResult != null && abortResult.failureReason == null; + // Same authoritative gate as the inactive-schedule path above: only a landed + // abort (or an already-terminal, drained generation) may settle this occurrence. + stopped = isStopConfirmed(abortResult); } catch (error) { logger.warn('[ResumeAgentController] Failed to stop stale scheduled resume', error); } diff --git a/api/server/index.js b/api/server/index.js index 9203250d70..539836cef6 100644 --- a/api/server/index.js +++ b/api/server/index.js @@ -43,6 +43,7 @@ const { updateInterfacePermissions, configureMessageFilterRegexValidator, configureFileConfigRegexEngine, + createScheduleWriteGate, waitForKeyvRedisClient, } = require('@librechat/api'); const { connectDb, indexSync } = require('~/db'); @@ -86,12 +87,11 @@ const trusted_proxy = Number(TRUST_PROXY) || 1; /* trust first proxy by default const app = express(); let serverReady = false; -let schedulesReady = false; +/** @type {import('@librechat/api').ScheduleEngineState} */ +let scheduleEngineState = 'starting'; const SERVER_NOT_READY_CODE = 'SERVER_NOT_READY'; const CHAT_START_RETRY_AFTER_SECONDS = '1'; -const SCHEDULES_NOT_READY_CODE = 'SCHEDULES_NOT_READY'; -const SCHEDULE_ENGINE_OPTIONAL_METHODS = new Set(['GET', 'HEAD', 'OPTIONS', 'DELETE']); const rejectChatStartsUntilReady = (req, res, next) => { if (serverReady || req.method !== 'POST' || req.path === '/abort') { @@ -105,16 +105,10 @@ const rejectChatStartsUntilReady = (req, res, next) => { }); }; -const rejectScheduleWritesUntilReady = (req, res, next) => { - if (schedulesReady || SCHEDULE_ENGINE_OPTIONAL_METHODS.has(req.method)) { - return next(); - } - res.set('Retry-After', CHAT_START_RETRY_AFTER_SECONDS); - return res.status(503).json({ - code: SCHEDULES_NOT_READY_CODE, - error: 'Scheduler is still starting. Please retry shortly.', - }); -}; +const rejectScheduleWritesUntilReady = createScheduleWriteGate({ + getState: () => scheduleEngineState, + retryAfterSeconds: CHAT_START_RETRY_AFTER_SECONDS, +}); const configureGenerationStreams = () => { const streamServices = createStreamServices(); @@ -419,9 +413,16 @@ const startServer = async () => { memoryDiagnostics.start(); } await initializeAgentTriggerService({ address: server.address() }); - schedulesReady = (await initializeScheduleEngine()) != null; - if (!schedulesReady) { - logger.warn('[schedules] write routes remain unavailable because the engine did not arm.'); + const scheduleEngineArmed = (await initializeScheduleEngine()) != null; + scheduleEngineState = scheduleEngineArmed ? 'armed' : 'unavailable'; + if (!scheduleEngineArmed) { + // Terminal, not transient: arming is attempted once, so schedule writes are refused + // for the life of this process. Logged at error level because the only other signal + // an operator gets is a 503 on every write — every other health signal stays green. + logger.error( + '[schedules] write routes are PERMANENTLY unavailable in this process: the engine did not arm. ' + + 'Resolve the cause logged above and restart.', + ); } serverReady = true; logger.info('Server readiness checks passing.'); diff --git a/packages/api/src/schedules/index.ts b/packages/api/src/schedules/index.ts index 4abce8246d..bc2755a66a 100644 --- a/packages/api/src/schedules/index.ts +++ b/packages/api/src/schedules/index.ts @@ -4,5 +4,6 @@ export * from './engine'; export * from './erasure'; export * from './fire'; export * from './handlers'; +export * from './readiness'; export * from './trigger'; export * from './types'; diff --git a/packages/api/src/schedules/readiness.spec.ts b/packages/api/src/schedules/readiness.spec.ts new file mode 100644 index 0000000000..9e3f358210 --- /dev/null +++ b/packages/api/src/schedules/readiness.spec.ts @@ -0,0 +1,92 @@ +import type { Response } from 'express'; +import type { ScheduleEngineState } from './readiness'; +import { + createScheduleWriteGate, + SCHEDULES_NOT_READY_CODE, + SCHEDULES_UNAVAILABLE_CODE, +} from './readiness'; + +function makeRes() { + const res = { + statusCode: 0, + body: undefined as unknown, + headers: {} as Record, + set(name: string, value: string) { + res.headers[name] = value; + return res; + }, + status(code: number) { + res.statusCode = code; + return res; + }, + json(payload: unknown) { + res.body = payload; + return res; + }, + }; + return res; +} + +function run(state: ScheduleEngineState, method: string) { + const res = makeRes(); + const next = jest.fn(); + createScheduleWriteGate({ getState: () => state, retryAfterSeconds: '1' })( + { method }, + res as unknown as Response, + next, + ); + return { res, next }; +} + +describe('createScheduleWriteGate', () => { + it('passes writes through once the engine is armed', () => { + const { res, next } = run('armed', 'POST'); + expect(next).toHaveBeenCalled(); + expect(res.statusCode).toBe(0); + }); + + it.each(['GET', 'HEAD', 'OPTIONS', 'DELETE'])( + 'never blocks %s, which does not need the engine', + (method) => { + for (const state of ['starting', 'unavailable'] as ScheduleEngineState[]) { + const { res, next } = run(state, method); + expect(next).toHaveBeenCalled(); + expect(res.statusCode).toBe(0); + } + }, + ); + + it('advertises a retry only while arming is genuinely still pending', () => { + const { res, next } = run('starting', 'POST'); + expect(next).not.toHaveBeenCalled(); + expect(res.statusCode).toBe(503); + expect(res.headers['Retry-After']).toBe('1'); + expect(res.body).toMatchObject({ code: SCHEDULES_NOT_READY_CODE }); + }); + + it('refuses terminally, without Retry-After, once arming has failed', () => { + const { res, next } = run('unavailable', 'POST'); + expect(next).not.toHaveBeenCalled(); + expect(res.statusCode).toBe(503); + // Nothing re-attempts arming, so a client obeying Retry-After here would poll a + // condition that cannot change without operator action. + expect(res.headers['Retry-After']).toBeUndefined(); + expect(res.body).toMatchObject({ code: SCHEDULES_UNAVAILABLE_CODE }); + }); + + it('re-reads the state on every request rather than capturing it at construction', () => { + let state: ScheduleEngineState = 'starting'; + const gate = createScheduleWriteGate({ getState: () => state, retryAfterSeconds: '1' }); + + const blocked = makeRes(); + gate({ method: 'POST' }, blocked as unknown as Response, jest.fn()); + expect(blocked.statusCode).toBe(503); + + state = 'armed'; + const allowed = makeRes(); + const next = jest.fn(); + gate({ method: 'POST' }, allowed as unknown as Response, next); + expect(next).toHaveBeenCalled(); + expect(allowed.statusCode).toBe(0); + }); +}); diff --git a/packages/api/src/schedules/readiness.ts b/packages/api/src/schedules/readiness.ts new file mode 100644 index 0000000000..3f5cc36f94 --- /dev/null +++ b/packages/api/src/schedules/readiness.ts @@ -0,0 +1,66 @@ +import type { Response, NextFunction } from 'express'; + +export const SCHEDULES_NOT_READY_CODE = 'SCHEDULES_NOT_READY'; +export const SCHEDULES_UNAVAILABLE_CODE = 'SCHEDULES_UNAVAILABLE'; + +/** + * Whether the schedule engine has been armed for this process. + * + * `starting` and `unavailable` both refuse writes, but they are not the same condition: + * arming is attempted EXACTLY ONCE at boot, so a failed arm is terminal for the life of + * the process. Collapsing the two into a single flag is what let a permanent outage be + * advertised with `Retry-After`. + */ +export type ScheduleEngineState = 'starting' | 'armed' | 'unavailable'; + +/** + * Reads and deletes never touch the engine: listing schedules, and removing one so it can + * no longer fire, must keep working even where nothing is armed. + */ +const ENGINE_OPTIONAL_METHODS = new Set(['GET', 'HEAD', 'OPTIONS', 'DELETE']); + +export interface ScheduleWriteGateOptions { + getState: () => ScheduleEngineState; + /** `Retry-After` for the genuinely transient window only. */ + retryAfterSeconds: string; +} + +export type ScheduleWriteGate = ( + req: { method: string }, + res: Response, + next: NextFunction, +) => Response | void; + +/** + * Guards schedule writes on engine readiness, answering with the retry contract that + * matches the real state: retry while arming is still pending, and a terminal refusal once + * it has definitively failed — nothing re-attempts arming, so a client obeying + * `Retry-After` there would poll a condition that cannot change without operator action. + */ +export function createScheduleWriteGate({ + getState, + retryAfterSeconds, +}: ScheduleWriteGateOptions): ScheduleWriteGate { + return function rejectScheduleWritesUntilReady( + req: { method: string }, + res: Response, + next: NextFunction, + ): Response | void { + const state = getState(); + if (state === 'armed' || ENGINE_OPTIONAL_METHODS.has(req.method)) { + return next(); + } + if (state === 'starting') { + res.set('Retry-After', retryAfterSeconds); + return res.status(503).json({ + code: SCHEDULES_NOT_READY_CODE, + error: 'Scheduler is still starting. Please retry shortly.', + }); + } + return res.status(503).json({ + code: SCHEDULES_UNAVAILABLE_CODE, + error: + 'Scheduler is unavailable in this deployment. Retrying will not help — check the server logs and resolve the startup failure.', + }); + }; +} diff --git a/packages/api/src/schedules/service.spec.ts b/packages/api/src/schedules/service.spec.ts index efc00f97b7..1c609baad0 100644 --- a/packages/api/src/schedules/service.spec.ts +++ b/packages/api/src/schedules/service.spec.ts @@ -1584,7 +1584,7 @@ describe('provider-drained schedule aborts', () => { })), } as unknown as typeof mockJobStore; const manager = jest.requireMock('../stream/GenerationJobManager').GenerationJobManager; - manager.abortJob = jest.fn(async () => ({ success: false })); + manager.abortJob = jest.fn(async () => ({ success: false, failureReason: 'already_settled' })); const delivered = await service.engineDeps.abortScheduledJob( 'c1', @@ -1598,6 +1598,35 @@ describe('provider-drained schedule aborts', () => { expect(delivered).toBe(true); }); + it.each(['generation_replaced', 'job_still_active', 'job_not_found'] as const)( + 'reports %s as an undelivered abort', + async (failureReason) => { + const service = makeService(jest.fn, [string]>().mockResolvedValue([])); + const deleteJob = jest.fn(async () => true); + mockJobStore = { + getJob: jest.fn(async () => ({ + status: 'running', + createdAt: 7, + scheduleId: 's1', + scheduledFor: '2026-01-01T00:00:00.000Z', + })), + deleteJob, + } as unknown as typeof mockJobStore; + const manager = jest.requireMock('../stream/GenerationJobManager').GenerationJobManager; + manager.abortJob = jest.fn(async () => ({ success: false, failureReason })); + + const delivered = await service.engineDeps.abortScheduledJob( + 'c1', + { scheduleId: 's1', scheduledFor: '2026-01-01T00:00:00.000Z' }, + { preserve: false }, + ); + + expect(delivered).toBe(false); + // Never destroy evidence for a generation this call did not stop. + expect(deleteJob).not.toHaveBeenCalled(); + }, + ); + it('deletes terminal evidence only after the exact provider drain is confirmed', async () => { const service = makeService(jest.fn, [string]>().mockResolvedValue([])); const deleteJob = jest.fn(async () => true); @@ -1611,7 +1640,7 @@ describe('provider-drained schedule aborts', () => { deleteJob, } as unknown as typeof mockJobStore; const manager = jest.requireMock('../stream/GenerationJobManager').GenerationJobManager; - manager.abortJob = jest.fn(async () => ({ success: false })); + manager.abortJob = jest.fn(async () => ({ success: false, failureReason: 'already_settled' })); const delivered = await service.engineDeps.abortScheduledJob( 'c1', diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index 393a724228..6aa16fd256 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -24,6 +24,7 @@ import { import { deleteAgentCheckpoint, captureAgentCheckpointGeneration } from '../agents/checkpointer'; import { fireSchedule, BALANCE_SKIP_DISABLE_THRESHOLD } from './fire'; import { GenerationJobManager } from '../stream/GenerationJobManager'; +import { isStopConfirmed } from '../stream/interfaces/IJobStore'; import { buildBalanceUpdateFields } from '../middleware/balance'; import { getAppConfigOptionsFromUser } from '../app/service'; import { isShutdownInProgress } from '../app/shutdown'; @@ -568,10 +569,9 @@ export function createSchedulesService( expectedCreatedAt: job.createdAt, awaitProviderDrain: true, }); - if ( - aborted.failureReason === 'generation_replaced' || - aborted.failureReason === 'job_still_active' - ) { + // Terminal-and-drained counts as delivered (see above); a replacement, a still-live + // run, or a job that vanished before the transition does not. + if (!isStopConfirmed(aborted)) { return false; } if (options?.preserve === false) { diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index 4b40b6c205..3a89dfb2d9 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -270,6 +270,28 @@ function isRecoverableTakeoverSplit( ); } +/** + * Name the reason a terminal-state CAS was lost, off the job that now holds the + * conversation. Losing to natural completion IS a stop; losing to a replacement, or to + * deletion, is not — and callers settle durable state on that distinction. + */ +function classifyLostAbortRace( + jobStillActive: boolean, + currentJob: SerializableJobData | null, + abortedCreatedAt: number, +): NonNullable { + if (jobStillActive) { + return 'job_still_active'; + } + if (currentJob == null) { + return 'job_not_found'; + } + if (currentJob.createdAt !== abortedCreatedAt) { + return 'generation_replaced'; + } + return 'already_settled'; +} + function buildTerminalPersistenceReconcile( job: Pick, ): t.FinalEvent { @@ -3709,6 +3731,10 @@ class GenerationJobManagerClass { content: [], jobData: null, success: false, + /** The job vanished between the caller's read and this one. No transition + * was made and no provider drain was awaited, so this says nothing about + * whether trailing owner work is still in flight. */ + failureReason: 'job_not_found', finalEvent: null, collectedUsage: [], }; @@ -3726,6 +3752,10 @@ class GenerationJobManagerClass { content: [], jobData: unlockedJob, success: false, + /** The pause never unlocked for THIS generation: the job was either deleted + * outright or a replacement took the conversation. A replacement is another + * run's state — settling or pruning on it would destroy the successor. */ + failureReason: unlockedJob == null ? 'job_not_found' : 'generation_replaced', finalEvent: null, collectedUsage: [], }; @@ -3748,6 +3778,10 @@ class GenerationJobManagerClass { content: [], jobData, success: false, + /** No transition was needed: the generation is already terminal, and the + * drain above (when requested) proves its provider segment can no longer + * persist. This is a stop, just not one this call made. */ + failureReason: 'already_settled', finalEvent: null, collectedUsage: [], }; @@ -3845,13 +3879,9 @@ class GenerationJobManagerClass { } return { success: false, - ...(jobStillActive - ? { failureReason: 'job_still_active' as const } - : options?.expectedCreatedAt != null && - currentJob != null && - currentJob?.createdAt !== options.expectedCreatedAt && { - failureReason: 'generation_replaced' as const, - }), + /** The drain above already ran when the caller required one, so an + * `already_settled` verdict here is a fully drained generation. */ + failureReason: classifyLostAbortRace(jobStillActive, currentJob, jobData.createdAt), jobData, content: abortContent, finalEvent: null, diff --git a/packages/api/src/stream/__tests__/abortFailureReason.spec.ts b/packages/api/src/stream/__tests__/abortFailureReason.spec.ts new file mode 100644 index 0000000000..2a283cf1b5 --- /dev/null +++ b/packages/api/src/stream/__tests__/abortFailureReason.spec.ts @@ -0,0 +1,132 @@ +/** + * Every `success: false` abort must name WHY it failed. Callers that settle durable + * state on an abort (schedule outcomes, checkpoint pruning) previously inferred a + * confirmed stop from the ABSENCE of a failure reason, which silently swept in the + * unlabeled not-found and already-terminal paths. + */ +import type { AbortResult } from '../interfaces/IJobStore'; +import { isStopConfirmed } from '../interfaces/IJobStore'; + +/** Suppress winston Console transport output (survives jest.resetModules) */ +jest.spyOn(console, 'log').mockImplementation(); + +async function configureManager() { + const { GenerationJobManager } = await import('../GenerationJobManager'); + const { InMemoryJobStore } = await import('../implementations/InMemoryJobStore'); + const { InMemoryEventTransport } = await import('../implementations/InMemoryEventTransport'); + + GenerationJobManager.configure({ + jobStore: new InMemoryJobStore(), + eventTransport: new InMemoryEventTransport(), + isRedis: false, + cleanupOnComplete: false, + }); + GenerationJobManager.initialize(); + return GenerationJobManager; +} + +describe('abortJob failure reasons', () => { + beforeEach(() => { + jest.resetModules(); + }); + + it('reports job_not_found when nothing occupies the stream', async () => { + const manager = await configureManager(); + + const result = await manager.abortJob('never-created'); + + expect(result.success).toBe(false); + expect(result.failureReason).toBe('job_not_found'); + expect(isStopConfirmed(result)).toBe(false); + + await manager.destroy(); + }); + + it('reports already_settled for a generation that is terminal before the call', async () => { + const manager = await configureManager(); + const streamId = 'abort-twice'; + await manager.createJob(streamId, 'user-1'); + + const first = await manager.abortJob(streamId); + expect(first.success).toBe(true); + + const second = await manager.abortJob(streamId); + + expect(second.success).toBe(false); + expect(second.failureReason).toBe('already_settled'); + // No transition was needed — the generation is already stopped, so a caller may + // still settle on it. This is the ONE failure reason that confirms a stop. + expect(isStopConfirmed(second)).toBe(true); + + await manager.destroy(); + }); + + it('reports generation_replaced when the epoch fence rejects a stale abort', async () => { + const manager = await configureManager(); + const streamId = 'epoch-fenced'; + const job = await manager.createJob(streamId, 'user-1'); + + const result = await manager.abortJob(streamId, { + expectedCreatedAt: job.createdAt - 1, + }); + + expect(result.success).toBe(false); + expect(result.failureReason).toBe('generation_replaced'); + expect(isStopConfirmed(result)).toBe(false); + + await manager.destroy(); + }); + + it('leaves no unlabeled failure across the reachable abort outcomes', async () => { + const manager = await configureManager(); + const streamId = 'labeled'; + const job = await manager.createJob(streamId, 'user-1'); + + const results = [ + await manager.abortJob('missing'), + await manager.abortJob(streamId, { expectedCreatedAt: job.createdAt - 1 }), + await manager.abortJob(streamId), + await manager.abortJob(streamId), + ]; + + for (const result of results) { + expect(result.success === true || result.failureReason != null).toBe(true); + } + + await manager.destroy(); + }); +}); + +describe('isStopConfirmed', () => { + const base: AbortResult = { + success: false, + jobData: null, + content: [], + finalEvent: null, + text: '', + collectedUsage: [], + }; + + it('confirms a landed abort', () => { + expect(isStopConfirmed({ ...base, success: true })).toBe(true); + }); + + it('confirms an already-terminal generation', () => { + expect(isStopConfirmed({ ...base, failureReason: 'already_settled' })).toBe(true); + }); + + it.each(['generation_replaced', 'job_still_active', 'job_not_found'] as const)( + 'refuses to confirm %s', + (failureReason) => { + expect(isStopConfirmed({ ...base, failureReason })).toBe(false); + }, + ); + + it('refuses to confirm a bare failure with no reason', () => { + expect(isStopConfirmed(base)).toBe(false); + }); + + it.each([null, undefined])('refuses to confirm %p', (result) => { + expect(isStopConfirmed(result)).toBe(false); + }); +}); diff --git a/packages/api/src/stream/__tests__/startup.spec.ts b/packages/api/src/stream/__tests__/startup.spec.ts index 7c7b3da8d0..7298dd4026 100644 --- a/packages/api/src/stream/__tests__/startup.spec.ts +++ b/packages/api/src/stream/__tests__/startup.spec.ts @@ -2278,7 +2278,9 @@ describe('GenerationJobManager startup telemetry', () => { const result = await aborting; expect(result).toMatchObject({ success: false, finalEvent: null }); - expect(result.failureReason).toBeUndefined(); + // Deletion is named for what it is. The point of this test is that it is NOT + // reported as a replacement — nothing took the conversation over. + expect(result.failureReason).toBe('job_not_found'); expect(job.abortController.signal.aborted).toBe(true); } finally { releaseTransition?.(); diff --git a/packages/api/src/stream/index.ts b/packages/api/src/stream/index.ts index 05995fd142..6ef140e0ae 100644 --- a/packages/api/src/stream/index.ts +++ b/packages/api/src/stream/index.ts @@ -28,6 +28,9 @@ export { isPendingActionExpired, isPendingActionStale, } from './interfaces/IJobStore'; +// Canonical "did this generation actually stop?" predicate — shared by every caller +// that settles durable state on an abort's outcome. +export { isStopConfirmed } from './interfaces/IJobStore'; export { STEER_ENQUEUE_NOT_RUNNING, STEER_ENQUEUE_QUEUE_FULL, diff --git a/packages/api/src/stream/interfaces/IJobStore.ts b/packages/api/src/stream/interfaces/IJobStore.ts index 50cb847ebc..a03e28298d 100644 --- a/packages/api/src/stream/interfaces/IJobStore.ts +++ b/packages/api/src/stream/interfaces/IJobStore.ts @@ -608,10 +608,13 @@ export interface UsageMetadata { export interface AbortResult { /** Whether the abort was successful */ success: boolean; - /** Distinguishes an epoch-fenced abort from ordinary not-found/terminal - * failures so an HTTP caller can return RUN_REPLACED instead of silently - * reporting success for a newer generation it deliberately did not stop. */ - failureReason?: 'generation_replaced' | 'job_still_active'; + /** Why the abort did not land. EVERY `success: false` return carries one, so a + * caller can separate a generation it must not settle (`generation_replaced`, + * `job_not_found`) from one that is still live (`job_still_active`) and from one + * that had already reached a terminal state (`already_settled` — the provider has + * also drained when `awaitProviderDrain` was requested). The ABSENCE of this field + * is not a stop confirmation; use `isStopConfirmed`. */ + failureReason?: 'generation_replaced' | 'job_still_active' | 'job_not_found' | 'already_settled'; /** The generation was stopped, but the caller's required durable side * effects failed before normal FINAL publication. The manager emitted a * conservative reconciliation frame instead. */ @@ -630,6 +633,22 @@ export interface AbortResult { pendingSteers?: TPendingSteer[]; } +/** + * Canonical "did this generation actually stop?" predicate — one definition shared by + * every caller that settles durable state on the answer (schedule outcomes, checkpoint + * pruning, capacity release). + * + * A landed abort confirms the stop. So does `already_settled`: the generation reached a + * terminal state on its own and, when the caller asked for `awaitProviderDrain`, its + * provider segment has drained, so nothing can still write. Every OTHER failure leaves a + * generation that is either still live (`job_still_active`), owned by someone else + * (`generation_replaced`), or unobservable from here without a drain (`job_not_found`) — + * none of which may be settled on. + */ +export function isStopConfirmed(result: AbortResult | null | undefined): boolean { + return result != null && (result.success === true || result.failureReason === 'already_settled'); +} + /** * Resume state for reconnecting clients */