🧷 fix: Finalize Stopped Compactions and Verify Their Anchor

api/server/controllers/agents/__tests__/request.partialDisconnect.spec.js

@@ -57,6 +57,10 @@ jest.mock('@librechat/api', () => ({
   resolveResumableRetention: jest.requireActual('@librechat/api').resolveResumableRetention,
   markAbortedCompactionContent: (...args) =>
     jest.requireActual('@librechat/api').markAbortedCompactionContent(...args),
+  resolveDisconnectSnapshotMode: (...args) =>
+    jest.requireActual('@librechat/api').resolveDisconnectSnapshotMode(...args),
+  settleExistingRowsBeforeErrorTurn: (...args) =>
+    jest.requireActual('@librechat/api').settleExistingRowsBeforeErrorTurn(...args),
   sendEvent: jest.fn(),
   persistedReasoningOverrideFields:
     jest.requireActual('@librechat/api').persistedReasoningOverrideFields,
@@ -342,4 +346,16 @@ describe('ResumableAgentController tenant context', () => {
     });
     expect(savedMessage.content).toHaveLength(1);
   });
+  /** The settling path (completion, error, abort) owns the final row: a
+   *  disconnect snapshot landing after it would reopen the settled turn as
+   *  an unfinished response. */
+  it('skips the partial save when the job record has settled', async () => {
+    await firePartialDisconnect(
+      { id: 'user-123' },
+      { createdAt: 1000, status: 'error' },
+      { aggregatedContent: [{ type: 'text', text: 'Partial response' }] },
+    );
+
+    expect(mockSaveMessage).not.toHaveBeenCalled();
+  });
 });

api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js

@@ -269,6 +269,10 @@ jest.mock('@librechat/api', () => ({
   resolveResumableRetention: jest.requireActual('@librechat/api').resolveResumableRetention,
   markAbortedCompactionContent: (...args) =>
     jest.requireActual('@librechat/api').markAbortedCompactionContent(...args),
+  resolveDisconnectSnapshotMode: (...args) =>
+    jest.requireActual('@librechat/api').resolveDisconnectSnapshotMode(...args),
+  settleExistingRowsBeforeErrorTurn: (...args) =>
+    jest.requireActual('@librechat/api').settleExistingRowsBeforeErrorTurn(...args),
   sendEvent: jest.fn(),
   /** Real, because whether a skipped-persistence turn may raise an indicator is under test. */
   isAnnounceableReply: jest.requireActual('@librechat/api').isAnnounceableReply,

api/server/controllers/agents/request.js

@@ -63,6 +63,8 @@ const {
   stampPreliminaryPrivateTextMessage,
   announceReply,
   announceErrorTurn,
+  settleExistingRowsBeforeErrorTurn,
+  resolveDisconnectSnapshotMode,
   markAbortedCompactionContent,
 } = require('@librechat/api');
 const { disposeClient } = require('~/server/cleanup');
@@ -458,23 +460,6 @@ async function saveErrorTurn(
     }
 
     const userId = req.user.id;
-    const existing = await getMessages(
-      { user: userId, messageId: errorMessageId, conversationId },
-      '_id',
-    );
-    if (existing.length > 0) {
-      return;
-    }
-    if (liveResponseMessageId != null && liveResponseMessageId !== errorMessageId) {
-      const partial = await getMessages(
-        { user: userId, messageId: liveResponseMessageId, conversationId },
-        '_id',
-      );
-      if (partial.length > 0) {
-        return;
-      }
-    }
-
     const reqCtx = {
       userId,
       isTemporary:
@@ -485,6 +470,35 @@ async function saveErrorTurn(
         req?._agentEventBindingRetention?.expiredAt ?? req?.resolvedConversation?.expiredAt,
       interfaceConfig: req?.config?.interfaceConfig,
     };
+    /** The existing-row settlement (which row a failed turn settles, and
+     *  whether its error row may be written at all) lives in @librechat/api;
+     *  this supplies the caller's reads and write. */
+    const coveredByExistingRow = await settleExistingRowsBeforeErrorTurn(req.body, {
+      userId,
+      conversationId,
+      errorMessageId,
+      liveResponseMessageId,
+      getMessages,
+      saveFinalizedTurn: (message) =>
+        saveMessage(reqCtx, message, {
+          context: 'api/server/controllers/agents/request.js - finalize failed compaction turn',
+        }),
+      announceSettledTurn: (messageId) =>
+        announceErrorTurn(
+          { stampConvoLastResponse },
+          {
+            userId,
+            conversationId,
+            messageId,
+            isTemporary: reqCtx.isTemporary,
+            context: 'AgentController - finalized failed compaction turn',
+          },
+        ),
+    });
+    if (coveredByExistingRow) {
+      return;
+    }
+
     const context = 'api/server/controllers/agents/request.js - failed turn';
     const endpoint = endpointOption?.endpoint;
     const model = getAgentResponseModel(req, endpointOption);
@@ -596,6 +610,13 @@ async function saveErrorTurn(
   }
 }
 
+/**
+ * The disconnect save is marker-only while the run is still live; a failed
+ * turn is what settles it, so a compaction's partial row is finalized here
+ * with the terminal outcome instead of keeping the snapshot's live-run
+ * marking. The decision lives in @librechat/api; this is the wiring, reusing
+ * the row the caller already loaded.
+ */
 function classifyScheduledFailure(error, aborted = false) {
   if (aborted || error?.code === 'SCHEDULE_NO_LONGER_ACTIVE') {
     return { status: 'interrupted', error: error?.message };
@@ -1937,6 +1958,10 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
        * record is the source, since the client-facing resume snapshot never
        * carries server-private state. */
       const contextMeta = jobRecord?.createdAt === jobCreatedAt ? jobRecord.contextMeta : undefined;
+      if (resolveDisconnectSnapshotMode(jobRecord, jobCreatedAt) === 'skip') {
+        logger.debug('[ResumableAgentController] Skipping partial response save for a settled job');
+        return;
+      }
 
       try {
         const partialMessage = {

api/server/routes/agents/__tests__/abort.spec.js

@@ -27,6 +27,7 @@ const mockSaveMessage = jest.fn();
 const mockHasPersistedPrivateText = jest.fn();
 const mockGetPrivateMessageTexts = jest.fn();
 const mockSaveConvo = jest.fn();
+const mockGetMessages = jest.fn(async () => [{ _id: 'existing-anchor' }]);
 
 const mockRecordScheduleOutcome = jest.fn();
 const mockBeginScheduledStop = jest.fn();
@@ -55,6 +56,7 @@ jest.mock('~/models', () => ({
     (await mockHasPersistedPrivateText(...args)) ? 'protected-row-id' : null,
   getPrivateMessageTexts: (...args) => mockGetPrivateMessageTexts(...args),
   saveConvo: (...args) => mockSaveConvo(...args),
+  getMessages: (...args) => mockGetMessages(...args),
 }));
 
 jest.mock('~/server/services/Schedules', () => ({
@@ -437,12 +439,72 @@ describe('Agent Abort Endpoint', () => {
           expect.objectContaining({
             messageId: compactionRowId,
             parentMessageId: anchorId,
-            unfinished: true,
+            unfinished: false,
             isCreatedByUser: false,
           }),
           expect.objectContaining({ context: expect.stringContaining('abort endpoint') }),
         );
       });
+
+      /** Stop can win the race before the branch loaded, so the projected
+       *  anchor names a row that was never written: a response persisted
+       *  there would be orphaned on reload. */
+      it('persists nothing when the compaction anchor was never written', async () => {
+        mockGetMessages.mockResolvedValueOnce([]);
+        const jobStreamId = 'test-stream-compact-unanchored';
+
+        mockGenerationJobManager.getJob.mockResolvedValue({
+          metadata: { userId: 'test-user-123', generationProtocolVersion: 2 },
+        });
+
+        const abortResult = {
+          success: true,
+          jobData: {
+            compact: true,
+            createdEventEmitted: true,
+            userMessage: {
+              messageId: 'never-persisted-leaf',
+              parentMessageId: 'older-response',
+              conversationId: jobStreamId,
+              text: '',
+            },
+            responseMessageId: 'compaction-response-2',
+            conversationId: jobStreamId,
+            endpoint: 'agents',
+            sender: 'TestAgent',
+            model: 'agent-1',
+          },
+          content: [
+            {
+              type: 'error',
+              error: JSON.stringify({ type: 'compaction_failed' }),
+              initiatedBy: 'user',
+            },
+          ],
+          text: '',
+        };
+        let beforePublishError;
+        mockGenerationJobManager.abortJob.mockImplementation(async (_streamId, options) => {
+          try {
+            await options.beforePublish(abortResult);
+          } catch (error) {
+            /** The manager catches this failure and publishes a
+             *  reconciliation frame instead of the normal FINAL. */
+            beforePublishError = error;
+          }
+          return abortResult;
+        });
+
+        const response = await request(app)
+          .post('/api/agents/chat/abort')
+          .set('X-LibreChat-Generation-Protocol', '2')
+          .send({ conversationId: jobStreamId, generationProtocolVersion: 2 });
+
+        expect(response.status).toBe(200);
+        expect(mockSaveMessage).not.toHaveBeenCalled();
+        expect(beforePublishError).toBeInstanceOf(Error);
+        expect(beforePublishError.message).toContain('anchor unavailable');
+      });
     });
 
     describe('Partial Response Saving', () => {

api/server/routes/agents/index.js

@@ -6,6 +6,7 @@ const {
   TERMINAL_PUBLICATION_RECONNECT_ERROR,
   hasPersistableAbortContent,
   announceStoppedReply,
+  resolveAbortedTurnPersistence,
   buildAbortedResponseMetadata,
   isPendingActionStale,
   toClientPendingAction,
@@ -56,6 +57,7 @@ const {
 } = require('~/server/controllers/agents/protocol');
 const {
   getFiles,
+  getMessages,
   saveMessage,
   saveConvo,
   getPersistedPrivateTextId,
@@ -789,15 +791,21 @@ router.post('/chat/abort', chatConfigMiddleware, async (req, res, next) => {
            * its parent and the preliminary-parent fence correctly rejects it. */
           const shouldPersistAbortedTurn =
             hasPersistableAbortContent(content) || jobData?.createdEventEmitted === true;
-          /** A compaction's `userMessage` is the already-persisted leaf
-           *  projected for identity only; upserting it would erase a user
-           *  leaf's text or turn an assistant leaf into an empty user row. */
-          const shouldPersistAnchor = jobData?.compact !== true;
+          /** The stopped turn's persistence plan (which rows to write, and
+           *  whether the normal FINAL must be withheld for a reconciliation
+           *  frame instead) comes from @librechat/api, decided from the
+           *  compaction anchor this route reads. */
+          const abortPersistencePlan = await resolveAbortedTurnPersistence(
+            jobData,
+            shouldPersistAbortedTurn,
+            { userId: req?.user?.id, getMessages },
+          );
+          persistenceErrors.push(...abortPersistencePlan.persistenceErrors);
 
           if (
             jobData?.userMessage?.messageId &&
             jobData?.responseMessageId &&
-            shouldPersistAbortedTurn
+            abortPersistencePlan.writeResponseRow
           ) {
             const messageContext = {
               userId: req?.user?.id,
@@ -827,7 +835,7 @@ router.post('/chat/abort', chatConfigMiddleware, async (req, res, next) => {
               endpoint: jobData.endpoint,
               iconURL: jobData.iconURL,
               model: jobData.model,
-              unfinished: true,
+              unfinished: abortPersistencePlan.responseUnfinished,
               error: false,
               isCreatedByUser: false,
               ...(Array.isArray(jobData.userSubmittedPaths) &&
@@ -859,7 +867,7 @@ router.post('/chat/abort', chatConfigMiddleware, async (req, res, next) => {
              * operation gets a chance to succeed. A compaction skips the
              * prerequisite: its anchor is the persisted leaf itself. */
             let persistedRequestId;
-            if (shouldPersistAnchor) {
+            if (abortPersistencePlan.writeUserRow) {
               try {
                 const persistedRequest = await saveAbortedUserMessage(
                   { saveMessage, getPersistedPrivateTextId, getPrivateMessageTexts },