fix: round sixteen: surgical index unwind, durable re-signal evidence, honest republication outcomes

- The cache-fill unwind removes only the failed fill's key from the
  user's reverse index instead of deleting the whole index, which left
  the user's OTHER live entries undiscoverable by later invalidations.
- abortJob skips its cleanup delete 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, and deleting it made
  every retry read a missing job and answer 404 while the peer-owned
  generation kept running. The completed-job TTL bounds the record.
- resignalAbort returns { delivered, published } instead of swallowing
  republication failures into an ownership boolean; the Stop route's
  terminal-status retry branch stays retryable (503 + Retry-After) when
  the signal provably never left this replica, and the scheduled path's
  evidence disposal keys on delivered as before.

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

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 });
         }

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();

packages/api/src/auth/userDocCache.ts

@@ -133,6 +133,28 @@ function sanitizeUserForCache(user: Partial<IUser>): 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<void> {
+  const indexKey = buildAuthUserDocReverseIndexKey(userId);
+  const existing = await store.get<string[]>(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.

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',

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

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<boolean> {
+  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,

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();