fix: close scheduled chat review follow-ups

client/src/data-provider/Schedules/__tests__/mutations.spec.tsx

@@ -0,0 +1,48 @@
+import { createElement } from 'react';
+import { QueryKeys } from 'librechat-data-provider';
+import { act, renderHook } from '@testing-library/react';
+import { QueryClient, QueryClientProvider } from '@tanstack/react-query';
+import type { ReactNode } from 'react';
+import { useDeleteScheduleMutation } from '../mutations';
+
+const mockDeleteSchedule = jest.fn();
+
+jest.mock('librechat-data-provider', () => {
+  const actual = jest.requireActual('librechat-data-provider');
+  return {
+    ...actual,
+    dataService: {
+      ...actual.dataService,
+      deleteSchedule: (...args: unknown[]) => mockDeleteSchedule(...args),
+    },
+  };
+});
+
+const createWrapper = (queryClient: QueryClient) =>
+  function Wrapper({ children }: { children: ReactNode }) {
+    return createElement(QueryClientProvider, { client: queryClient }, children);
+  };
+
+describe('useDeleteScheduleMutation', () => {
+  it('invalidates the schedules list when deletion returns an error after mutating state', async () => {
+    const queryClient = new QueryClient({
+      logger: { log: jest.fn(), warn: jest.fn(), error: jest.fn() },
+      defaultOptions: {
+        queries: { retry: false },
+        mutations: { retry: false },
+      },
+    });
+    const invalidateQueries = jest.spyOn(queryClient, 'invalidateQueries');
+    mockDeleteSchedule.mockRejectedValueOnce(new Error('Could not confirm active run stopped'));
+    const { result } = renderHook(() => useDeleteScheduleMutation(), {
+      wrapper: createWrapper(queryClient),
+    });
+
+    await act(async () => {
+      await result.current.mutateAsync('schedule-1').catch(() => undefined);
+    });
+
+    expect(invalidateQueries).toHaveBeenCalledWith([QueryKeys.schedules]);
+    queryClient.clear();
+  });
+});

client/src/data-provider/Schedules/mutations.ts

@@ -53,9 +53,12 @@ export const useDeleteScheduleMutation = (
     (id: string) => dataService.deleteSchedule(id),
     {
       ...options,
-      onSuccess: (...args) => {
+      // An `unconfirmed` 503 is still a server-side mutation: the schedule has
+      // already been disabled, marked deleting, and hidden from list reads.
+      // Refresh on either outcome so the card cannot remain actionable in cache.
+      onSettled: (...args) => {
         queryClient.invalidateQueries([QueryKeys.schedules]);
-        options?.onSuccess?.(...args);
+        options?.onSettled?.(...args);
       },
     },
   );

packages/api/src/stream/GenerationJobManager.ts

@@ -20,6 +20,7 @@ import type { StandardGraph } from '@librechat/agents';
 import type {
   SerializableJobData,
   CreatedJobData,
+  ReplacedGeneration,
   IEventTransport,
   UsageMetadata,
   AbortResult,
@@ -715,6 +716,14 @@ interface PreparedSubscription {
   deferDeliveryUntilActivated: boolean;
 }
 
+interface CommittedCreateRecovery {
+  job: CreatedJobData;
+  /** Transaction receipts already delivered during ambiguous-create recovery.
+   * They may be absent from the repaired job after their fenced ACK, but the
+   * caller still needs them to classify its stale pre-create snapshot. */
+  handledPredecessors: readonly ReplacedGeneration[];
+}
+
 type DeferredDelivery =
   | { type: 'chunk'; event: t.ServerSentEvent }
   | { type: 'done'; event: t.ServerSentEvent }
@@ -1910,7 +1919,7 @@ class GenerationJobManagerClass {
     tenantId: string | undefined,
     options: CreateGenerationJobOptions,
     creationAttemptId: string,
-  ): Promise<CreatedJobData | null> {
+  ): Promise<CommittedCreateRecovery | null> {
     const clientRequestId = options.idempotencyClientRequestId;
     const claimToken = options.idempotencyClaimToken;
     const expectedConversationId = conversationId ?? streamId;
@@ -1924,6 +1933,7 @@ class GenerationJobManagerClass {
       (job.idempotencyClientRequestId ?? undefined) === clientRequestId;
 
     let committedJob = (await this.jobStore.getJob(streamId)) as CreatedJobData | null;
+    let handledPredecessors: readonly ReplacedGeneration[] = [];
     if (
       committedJob != null &&
       committedJob.userId === userId &&
@@ -1933,6 +1943,9 @@ class GenerationJobManagerClass {
       // mutable because later recovery may refresh it, so TypeScript cannot
       // safely retain a narrowing on that outer binding across the await.
       const currentCommittedJob = committedJob;
+      handledPredecessors =
+        currentCommittedJob.replacedJobs ??
+        (currentCommittedJob.replacedJob != null ? [currentCommittedJob.replacedJob] : []);
       const ownsCommittedCreate = matchesCommittedCreate(currentCommittedJob);
       const predecessorsDelivered = await this.processDurableReplacementReceipts(
         streamId,
@@ -2048,7 +2061,7 @@ class GenerationJobManagerClass {
         if (repaired) {
           const confirmed = await this.jobStore.getJob(streamId);
           if (matchesCommittedCreate(confirmed)) {
-            return confirmed;
+            return { job: confirmed, handledPredecessors };
           }
           return null;
         }
@@ -2211,6 +2224,7 @@ class GenerationJobManagerClass {
     // so replacement must explicitly close/handoff that old attachment.
     const observedPredecessorJob = await this.jobStore.getJob(streamId);
     let jobData: CreatedJobData;
+    let recoveredPredecessors: readonly ReplacedGeneration[] = [];
     try {
       jobData = await this.jobStore.createJob(
         streamId,
@@ -2263,7 +2277,7 @@ class GenerationJobManagerClass {
         throw error;
       }
 
-      let recovered: CreatedJobData | null = null;
+      let recovered: CommittedCreateRecovery | null = null;
       try {
         recovered = await this.recoverCommittedCreate(
           streamId,
@@ -2282,7 +2296,8 @@ class GenerationJobManagerClass {
       if (recovered == null) {
         throw error;
       }
-      jobData = recovered;
+      jobData = recovered.job;
+      recoveredPredecessors = recovered.handledPredecessors;
     }
 
     if (options.idempotencyClientRequestId != null && options.idempotencyClaimToken != null) {
@@ -2397,14 +2412,23 @@ class GenerationJobManagerClass {
     }
 
     const replacementEpochs = new Map<number, string | undefined>();
-    const exactPredecessors =
+    const durableExactPredecessors =
       jobData.replacedJobs ?? (jobData.replacedJob != null ? [jobData.replacedJob] : []);
     const exactPredecessorsByEpoch = new Map(
-      exactPredecessors.map((predecessor) => [predecessor.createdAt, predecessor]),
+      [...recoveredPredecessors, ...durableExactPredecessors].map((predecessor) => [
+        predecessor.createdAt,
+        predecessor,
+      ]),
     );
+    const exactPredecessors = [...exactPredecessorsByEpoch.values()];
     const observedPredecessorReceipt = observedPredecessorJob
       ? exactPredecessorsByEpoch.get(observedPredecessorJob.createdAt)
       : undefined;
+    const observedPredecessorCoveredByLaterReceipt =
+      observedPredecessorJob != null &&
+      exactPredecessors.some(
+        (predecessor) => predecessor.createdAt > observedPredecessorJob.createdAt,
+      );
     if (
       observedPredecessorJob &&
       observedPredecessorJob.createdAt !== jobData.createdAt &&
@@ -2413,7 +2437,12 @@ class GenerationJobManagerClass {
       // The transaction-time predecessor is authoritative for the same
       // epoch. A pre-read may say running even though it terminalized before
       // replacement; do not publish a contradictory replaced/abort pair.
-      observedPredecessorReceipt == null
+      observedPredecessorReceipt == null &&
+      // A newer transaction-time receipt also proves this pre-read was already
+      // superseded before our create committed. If its earlier handoff were
+      // still unconfirmed, Redis would have inherited that epoch in the same
+      // retained receipt chain; its absence is the predecessor's fenced ACK.
+      !observedPredecessorCoveredByLaterReceipt
     ) {
       replacementEpochs.set(
         observedPredecessorJob.createdAt,

packages/api/src/stream/__tests__/abortSignalHonesty.spec.ts

@@ -284,6 +284,82 @@ describe('abort signal honesty across replicas', () => {
     await owner.destroy();
   });
 
+  it('does not let a predecessor owner release clear the current generation Stop fence', async () => {
+    const { GenerationJobManagerClass } = await import('../GenerationJobManager');
+    const { InMemoryJobStore } = await import('../implementations/InMemoryJobStore');
+    const { InMemoryEventTransport } = await import('../implementations/InMemoryEventTransport');
+    const jobStore = new InMemoryJobStore({ ttlAfterComplete: 0 });
+    const retainAbortDelivery = jest.fn(async () => undefined);
+    const clearAbortDelivery = jest.fn(async () => undefined);
+    const fallback = {
+      renew: jest.fn(async () => undefined),
+      clear: jest.fn(async () => undefined),
+      retainAbortDelivery,
+      clearAbortDelivery,
+    };
+    const owner = new GenerationJobManagerClass();
+    owner.configure({
+      jobStore,
+      eventTransport: new InMemoryEventTransport(),
+      isRedis: true,
+      userFinalizationFallback: fallback,
+    });
+    owner.initialize();
+    const peerTransport = Object.assign(new InMemoryEventTransport(), {
+      emitAbortConfirmed: jest.fn(async () => false),
+    });
+    const peer = new GenerationJobManagerClass();
+    peer.configure({
+      jobStore,
+      eventTransport: peerTransport,
+      isRedis: true,
+      userFinalizationFallback: fallback,
+    });
+    peer.initialize();
+
+    try {
+      const predecessor = await owner.createJob(
+        'conv-generation-fence',
+        'user-1',
+        'conv-generation-fence',
+      );
+      const predecessorAbort = await owner.abortJob('conv-generation-fence');
+      expect(predecessorAbort.signalDelivered).toBe(true);
+
+      const current = await owner.createJob(
+        'conv-generation-fence',
+        'user-1',
+        'conv-generation-fence',
+      );
+      expect(current.createdAt).toBeGreaterThan(predecessor.createdAt);
+      const currentAbort = await peer.abortJob('conv-generation-fence');
+      expect(currentAbort).toMatchObject({
+        success: true,
+        signalDelivered: false,
+        signalPublished: false,
+      });
+      expect(retainAbortDelivery).toHaveBeenLastCalledWith(
+        'user-1',
+        undefined,
+        'conv-generation-fence',
+      );
+      clearAbortDelivery.mockClear();
+
+      // Generation A still has its local owner hold while generation B's failed
+      // Stop owns the stream-level durable recovery fence. A's late catch must not
+      // clear B's evidence merely because A has a hold for the same stream.
+      await owner.releaseOwnerLease(
+        'user-1',
+        'conv-generation-fence',
+        undefined,
+        predecessor.createdAt,
+      );
+      expect(clearAbortDelivery).not.toHaveBeenCalled();
+    } finally {
+      await Promise.allSettled([peer.destroy(), owner.destroy()]);
+    }
+  });
+
   it('heartbeats the abort contender while required persistence exceeds the lease TTL', async () => {
     const { owner, jobStore } = await makeManagers();
     await owner.createJob('conv-slow-stop-persistence', 'user-1', 'conv-slow-stop-persistence');

packages/api/src/stream/__tests__/protocolRollout.stream_integration.spec.ts

@@ -278,15 +278,22 @@ describe('Redis generation protocol rollout bridge', () => {
   });
 
   test('recovers a lost create reply, repairs membership, and stops the exact raced predecessor', async () => {
-    const abortHandlers = new Map<string, Set<(generationId?: number) => void | boolean>>();
+    const abortHandlers = new Map<
+      string,
+      Set<(generationId?: number) => void | boolean | Promise<void | boolean>>
+    >();
     const acknowledgedAborts = new Map<string, Set<number>>();
+    const confirmedAbortRequests: number[] = [];
     const eventTransport = Object.assign(new InMemoryEventTransport(), {
       emitAbort(streamId: string, generationId?: number) {
         for (const handler of abortHandlers.get(streamId) ?? []) {
           handler(generationId);
         }
       },
-      async onAbort(streamId: string, handler: (generationId?: number) => void) {
+      async onAbort(
+        streamId: string,
+        handler: (generationId?: number) => void | boolean | Promise<void | boolean>,
+      ) {
         let handlers = abortHandlers.get(streamId);
         if (handlers == null) {
           handlers = new Set();
@@ -296,9 +303,10 @@ describe('Redis generation protocol rollout bridge', () => {
         return () => handlers?.delete(handler);
       },
       async emitAbortConfirmed(streamId: string, generationId: number) {
+        confirmedAbortRequests.push(generationId);
         let acknowledged = acknowledgedAborts.get(streamId)?.has(generationId) === true;
         for (const handler of abortHandlers.get(streamId) ?? []) {
-          acknowledged = handler(generationId) === true || acknowledged;
+          acknowledged = (await handler(generationId)) === true || acknowledged;
         }
         if (acknowledged) {
           let generations = acknowledgedAborts.get(streamId);
@@ -385,6 +393,8 @@ describe('Redis generation protocol rollout bridge', () => {
       expect(racedPredecessor!.createdAt).toBeGreaterThan(stalePredecessor.createdAt);
       expect(replacement.createdAt).toBeGreaterThan(racedPredecessor!.createdAt);
       expect(racedPredecessor!.abortController.signal.aborted).toBe(true);
+      expect(confirmedAbortRequests).toContain(racedPredecessor!.createdAt);
+      expect(confirmedAbortRequests).not.toContain(stalePredecessor.createdAt);
       expect(doneEvents).toContainEqual({
         event: expect.objectContaining({
           final: true,
@@ -409,7 +419,10 @@ describe('Redis generation protocol rollout bridge', () => {
   });
 
   test('a superseded lost-reply helper delivers but does not acknowledge the crashed successor chain', async () => {
-    const abortHandlers = new Map<string, Set<(generationId?: number) => void | boolean>>();
+    const abortHandlers = new Map<
+      string,
+      Set<(generationId?: number) => void | boolean | Promise<void | boolean>>
+    >();
     const acknowledgedAborts = new Map<string, Set<number>>();
     const eventTransport = Object.assign(new InMemoryEventTransport(), {
       emitAbort(streamId: string, generationId?: number) {
@@ -417,7 +430,10 @@ describe('Redis generation protocol rollout bridge', () => {
           handler(generationId);
         }
       },
-      async onAbort(streamId: string, handler: (generationId?: number) => void) {
+      async onAbort(
+        streamId: string,
+        handler: (generationId?: number) => void | boolean | Promise<void | boolean>,
+      ) {
         let handlers = abortHandlers.get(streamId);
         if (handlers == null) {
           handlers = new Set();
@@ -429,7 +445,7 @@ describe('Redis generation protocol rollout bridge', () => {
       async emitAbortConfirmed(streamId: string, generationId: number) {
         let acknowledged = acknowledgedAborts.get(streamId)?.has(generationId) === true;
         for (const handler of abortHandlers.get(streamId) ?? []) {
-          acknowledged = handler(generationId) === true || acknowledged;
+          acknowledged = (await handler(generationId)) === true || acknowledged;
         }
         if (acknowledged) {
           let generations = acknowledgedAborts.get(streamId);