diff --git a/api/server/routes/agents/__tests__/abort.spec.js b/api/server/routes/agents/__tests__/abort.spec.js index 13ab609468..8a2d64b20a 100644 --- a/api/server/routes/agents/__tests__/abort.spec.js +++ b/api/server/routes/agents/__tests__/abort.spec.js @@ -20,7 +20,7 @@ const mockLogger = { const mockGenerationJobManager = { getJob: jest.fn(), abortJob: jest.fn(), - resignalAbort: jest.fn(async () => false), + resignalAbort: jest.fn(async () => ({ delivered: false, published: true })), getActiveJobIdsForUser: jest.fn(), }; @@ -831,6 +831,27 @@ describe('Agent Abort Endpoint', () => { expect(mockSaveMessage).not.toHaveBeenCalled(); }); + it('stays retryable when the terminal-branch republication also fails', async () => { + mockGenerationJobManager.getJob.mockResolvedValue(interactiveJob); + mockGenerationJobManager.abortJob.mockResolvedValue({ + success: false, + content: [], + jobData: { status: 'aborted' }, + }); + // The retry's republish is ALSO swallowed on this replica: answering 200 + // here told the client the stop landed while the signal provably never left. + mockGenerationJobManager.resignalAbort.mockResolvedValueOnce({ + delivered: false, + published: false, + }); + + const response = await request(app) + .post('/api/agents/chat/abort') + .send({ conversationId: 'test-conv' }); + + expect(response.status).toBe(503); + }); + it('re-signals an already-aborted job instead of trusting terminal status', async () => { mockGenerationJobManager.getJob.mockResolvedValue(interactiveJob); // A previous Stop won the CAS; its publication may never have left that diff --git a/api/server/routes/agents/index.js b/api/server/routes/agents/index.js index b13665d6c3..c0595bc638 100644 --- a/api/server/routes/agents/index.js +++ b/api/server/routes/agents/index.js @@ -476,7 +476,7 @@ router.post('/chat/abort', configMiddleware, async (req, res) => { abortResult.signalDelivered === false && abortResult.signalPublished === false ) { - await GenerationJobManager.resignalAbort(jobStreamId, job.createdAt).catch(() => false); + await GenerationJobManager.resignalAbort(jobStreamId, job.createdAt).catch(() => undefined); res.set('Retry-After', '2'); return res.status(503).json({ error: 'Stop recorded but not yet delivered to the generation. Please retry.', @@ -495,9 +495,21 @@ router.post('/chat/abort', configMiddleware, async (req, res) => { // trust it (the interactive mirror of the scheduled path's resignalAbort), // so the retry after a failed-publish 503 actually redelivers. if (abortResult.jobData.status === 'aborted') { - await GenerationJobManager.resignalAbort(jobStreamId, job.createdAt).catch( - () => undefined, - ); + const resignal = await GenerationJobManager.resignalAbort( + jobStreamId, + job.createdAt, + ).catch(() => ({ delivered: false, published: false })); + // A swallowed republication failure must not read as success: with the + // signal provably still on this replica, the peer-owned generation keeps + // running, so the response stays retryable until a publish leaves (or + // this process turns out to own the generation). + if (!scheduledFireIdentity && !resignal.delivered && !resignal.published) { + res.set('Retry-After', '2'); + return res.status(503).json({ + error: 'Stop recorded but not yet delivered to the generation. Please retry.', + aborted: null, + }); + } await resolveStopAttempt(); return res.json({ success: true, aborted: jobStreamId }); } diff --git a/packages/api/src/auth/userDocCache.spec.ts b/packages/api/src/auth/userDocCache.spec.ts index 79c8b454e9..d8c10fad9c 100644 --- a/packages/api/src/auth/userDocCache.spec.ts +++ b/packages/api/src/auth/userDocCache.spec.ts @@ -209,6 +209,38 @@ describe('auth user document cache helpers', () => { expect(store.values.has(buildAuthUserDocReverseIndexKey(userId.toString()))).toBe(false); }); + it("preserves the user's OTHER cache entries when unwinding a failed fill", async () => { + const store = makeStore(); + const userId = new Types.ObjectId(); + const survivingKey = 'auth-user-doc:v1:other-session'; + const failingKey = 'auth-user-doc:v1:failing-fill'; + // Another live entry for the same user, already indexed. + store.values.set(survivingKey, { version: 1, cachedAt: Date.now(), user: {} }); + store.values.set(buildAuthUserDocReverseIndexKey(userId.toString()), [survivingKey]); + const realGet = store.get; + store.get = (async (key: string) => { + if (key.startsWith('auth-user-doc-tombstone:')) { + throw new Error('redis blip'); + } + return realGet(key); + }) as typeof store.get; + + await setCachedAuthUserDoc(store, failingKey, { + _id: userId, + id: userId.toString(), + email: 'user@example.com', + }); + + // Deleting the WHOLE index left the surviving entry undiscoverable: later + // mutations and deletions could no longer invalidate it, so a stale document + // was served until its TTL. Only the failed fill's key is removed. + expect(store.values.has(failingKey)).toBe(false); + expect(store.values.get(buildAuthUserDocReverseIndexKey(userId.toString()))).toEqual([ + survivingKey, + ]); + expect(store.values.has(survivingKey)).toBe(true); + }); + it('deduplicates reverse-index keys and caps the remembered set', async () => { const store = makeStore(); const objectId = new Types.ObjectId(); diff --git a/packages/api/src/auth/userDocCache.ts b/packages/api/src/auth/userDocCache.ts index 98dc45f57f..8933e7b74f 100644 --- a/packages/api/src/auth/userDocCache.ts +++ b/packages/api/src/auth/userDocCache.ts @@ -133,6 +133,28 @@ function sanitizeUserForCache(user: Partial): CachedAuthUser { return sanitized; } +/** Removes ONE cache key from the user's reverse index, preserving the rest: an + * unwind that deleted the whole index left the user's OTHER live entries + * undiscoverable, so later mutations and deletions could no longer invalidate + * them and a stale document survived to its TTL. */ +async function forgetUserCacheKey( + store: AuthUserDocCacheStore, + userId: string, + cacheKey: string, +): Promise { + const indexKey = buildAuthUserDocReverseIndexKey(userId); + const existing = await store.get(indexKey); + if (!Array.isArray(existing)) { + return; + } + const remaining = existing.filter((value) => value !== cacheKey); + if (remaining.length === 0) { + await store.delete(indexKey); + return; + } + await store.set(indexKey, remaining, AUTH_USER_DOC_CACHE_TTL_MS); +} + async function rememberUserCacheKey( store: AuthUserDocCacheStore, userId: string, @@ -201,7 +223,7 @@ export async function setCachedAuthUserDoc( const tombstoned = await store.get(buildAuthUserDocTombstoneKey(userId)); if (tombstoned != null) { await store.delete(cacheKey); - await store.delete(buildAuthUserDocReverseIndexKey(userId)); + await forgetUserCacheKey(store, userId, cacheKey); return 'tombstoned'; } } @@ -220,7 +242,7 @@ export async function setCachedAuthUserDoc( try { await store.delete(cacheKey); if (userId) { - await store.delete(buildAuthUserDocReverseIndexKey(userId)); + await forgetUserCacheKey(store, userId, cacheKey); } } catch { // TTL-bounded residual; nothing further to do. diff --git a/packages/api/src/schedules/service.spec.ts b/packages/api/src/schedules/service.spec.ts index 7b37565e10..b26be22919 100644 --- a/packages/api/src/schedules/service.spec.ts +++ b/packages/api/src/schedules/service.spec.ts @@ -24,7 +24,7 @@ jest.mock('../stream/GenerationJobManager', () => ({ // loop is driven purely by getActiveRunsForUser (the run rows). getJobStore: () => mockJobStore, abortJob: jest.fn(), - resignalAbort: jest.fn(async () => false), + resignalAbort: jest.fn(async () => ({ delivered: false, published: true })), isRedis: false, }, })); @@ -1171,7 +1171,7 @@ describe('abort retries re-signal instead of trusting terminal status', () => { })), } as unknown as typeof mockJobStore; const manager = jest.requireMock('../stream/GenerationJobManager').GenerationJobManager; - manager.resignalAbort = jest.fn(async () => false); + manager.resignalAbort = jest.fn(async () => ({ delivered: false, published: true })); // The first abort flipped the job before its publication; a retry that trusts // the terminal status returns "delivered" without republishing, and a failed @@ -1198,7 +1198,7 @@ describe('abort retries re-signal instead of trusting terminal status', () => { deleteJob, } as unknown as typeof mockJobStore; const manager = jest.requireMock('../stream/GenerationJobManager').GenerationJobManager; - manager.resignalAbort = jest.fn(async () => false); + manager.resignalAbort = jest.fn(async () => ({ delivered: false, published: true })); const delivered = await service.engineDeps.abortScheduledJob( 'c1', diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index 09a1a0e179..12306d09dc 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -491,20 +491,20 @@ export function createSchedulesService( // every retry instead; delivery stays ownership-honest, and the caller's // bounded drain confirms on the owner's settle once the signal actually lands. if (job.status === 'aborted') { - const delivered = await GenerationJobManager.resignalAbort( + const resignal = await GenerationJobManager.resignalAbort( conversationId, job.createdAt, ).catch((err) => { logger.warn('[schedules] failed to re-signal abort:', err); - return false; + return { delivered: false, published: false }; }); - if (options?.preserve === false && delivered) { + if (options?.preserve === false && resignal.delivered) { // Only provably-quiet evidence is disposable (account deletion hard-deletes // the run rows, so nothing would ever clear this job later). Undelivered: // keep it — the drain stays unconfirmed and a later pass re-signals. await store.deleteJob(conversationId, job.createdAt); } - return delivered; + return resignal.delivered; } // Finished naturally (`complete`/`error`): the generation persisted and stopped // on its own — nothing to signal. For a per-schedule delete (preserve) leave the diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index f000f67834..ce2aaeb635 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -1458,19 +1458,26 @@ class GenerationJobManagerClass { * OWNERSHIP, never publish success. Callers confirm actual delivery by the run * settling (the owner settles last), which their bounded drains already poll. */ - async resignalAbort(streamId: string, expectedCreatedAt?: number): Promise { + async resignalAbort( + streamId: string, + expectedCreatedAt?: number, + ): Promise<{ delivered: boolean; published: boolean }> { const jobData = await this.jobStore.getJob(streamId); if ( jobData == null || jobData.status !== 'aborted' || (expectedCreatedAt != null && jobData.createdAt !== expectedCreatedAt) ) { - return false; + return { delivered: false, published: false }; } const runtime = this.runtimeState.get(streamId); if (runtime?.createdAt === jobData.createdAt) { runtime.abortController.abort(); } + // `published` reports whether the republication left this replica (see + // AbortResult.signalPublished); callers must stay retryable when it did not, + // instead of discarding a swallowed failure and answering success. + let published = true; if (this.eventTransport.emitAbort) { try { await withTimeout( @@ -1479,10 +1486,11 @@ class GenerationJobManagerClass { `Abort republication timed out for ${streamId}`, ); } catch (err) { + published = false; logger.error(`[GenerationJobManager] Failed to republish abort for ${streamId}:`, err); } } - return this.ownedJobs.get(streamId) === jobData.createdAt; + return { delivered: this.ownedJobs.get(streamId) === jobData.createdAt, published }; } async abortJob( @@ -1778,7 +1786,12 @@ class GenerationJobManagerClass { if (runtime) { runtime.startupTelemetry = undefined; } - if (this._cleanupOnComplete && !options?.preserveForReconcile) { + // SKIPPED when the abort publication provably failed: the terminal job is the + // only thing a retry (or the route's immediate resignalAbort) can re-signal + // FROM — deleting it made every retry read a missing job, publish nothing, and + // answer 404 while the peer-owned generation kept running. The store's + // completed-job TTL bounds the retained record. + if (this._cleanupOnComplete && !options?.preserveForReconcile && abortSignalPublished) { // A replacement created after the abort CAS makes this a safe no-op. Best-effort // and bounded like every other post-CAS store/transport call: the job is already // terminal, so a leaked record falls to the store TTL / retained-job reaper, diff --git a/packages/api/src/stream/__tests__/steering.spec.ts b/packages/api/src/stream/__tests__/steering.spec.ts index 143d4dc076..13bc4d2956 100644 --- a/packages/api/src/stream/__tests__/steering.spec.ts +++ b/packages/api/src/stream/__tests__/steering.spec.ts @@ -512,6 +512,38 @@ describe('SteeringLifecycle via GenerationJobManager.steering (in-memory)', () = } }); + test('abortJob retains the job when the publication provably failed', async () => { + const streamId = 'steer-abort-publish-fails'; + const transport = new InMemoryEventTransport(); + (transport as IEventTransport).emitAbort = async () => { + throw new Error('redis publish failed'); + }; + const localManager = new GenerationJobManagerClass(); + localManager.configure({ + jobStore, + eventTransport: transport, + isRedis: false, + // Production default: terminal jobs are deleted on completion... + cleanupOnComplete: true, + }); + localManager.initialize(); + const job = await localManager.createJob(streamId, 'user-1'); + + try { + const result = await localManager.abortJob(streamId); + expect(result.success).toBe(true); + expect(result.signalPublished).toBe(false); + // ...but a failed publish RETAINS it: the terminal job is the only thing + // a retry (or the route's immediate resignalAbort) can re-signal from. + await expect(jobStore.getJob(streamId)).resolves.toMatchObject({ + status: 'aborted', + createdAt: job.createdAt, + }); + } finally { + await localManager.destroy(); + } + }); + test('abortJob publishes nothing when natural completion wins its terminal CAS', async () => { const streamId = 'steer-abort-loses-terminal-race'; const eventTransport = new InMemoryEventTransport();