📴 fix: Persist Cancelled Compactions With Their Identity

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

@@ -49,6 +49,15 @@ jest.mock('@librechat/data-schemas', () => ({
 jest.mock('@librechat/api', () => ({
   getAgentErrorMetadata: (...args) =>
     jest.requireActual('@librechat/api').getAgentErrorMetadata(...args),
+  markAbortedCompactionContent: (...args) =>
+    jest.requireActual('@librechat/api').markAbortedCompactionContent(...args),
+  isSettledJobRecord: (...args) => jest.requireActual('@librechat/api').isSettledJobRecord(...args),
+  resolveDisconnectSnapshotMode: (...args) =>
+    jest.requireActual('@librechat/api').resolveDisconnectSnapshotMode(...args),
+  resolveReconciledSnapshotEnvelope: (...args) =>
+    jest.requireActual('@librechat/api').resolveReconciledSnapshotEnvelope(...args),
+  settleExistingRowsBeforeErrorTurn: (...args) =>
+    jest.requireActual('@librechat/api').settleExistingRowsBeforeErrorTurn(...args),
   sendEvent: jest.fn(),
   isScheduleFireRequest: jest.fn(() => false),
   exemptFromConcurrencyLimiter: jest.fn(() => false),
@@ -163,6 +172,7 @@ describe('ResumableAgentController tenant context', () => {
   const firePartialDisconnect = async (
     user,
     jobRecord = { createdAt: 1000, contextMeta: partialContextMeta },
+    { body = {}, aggregatedContent = [{ type: 'text', text: 'Partial response' }] } = {},
   ) => {
     let allSubscribersLeftHandler;
     mockGenerationJobManager.getJobStore.mockReturnValue({
@@ -210,6 +220,7 @@ describe('ResumableAgentController tenant context', () => {
           endpoint: 'agents',
           modelOptions: { model: 'gpt-4.1' },
         },
+        ...body,
       },
       config: {},
     };
@@ -222,7 +233,7 @@ describe('ResumableAgentController tenant context', () => {
     await AgentController(req, res, jest.fn(), initializeClient, null);
     expect(allSubscribersLeftHandler).toEqual(expect.any(Function));
 
-    await allSubscribersLeftHandler([{ type: 'text', text: 'Partial response' }]);
+    await allSubscribersLeftHandler(aggregatedContent);
     return tenantSeenBySave;
   };
 
@@ -268,4 +279,180 @@ describe('ResumableAgentController tenant context', () => {
     expect(tenantSeenBySave).toBeUndefined();
     expect(mockSaveMessage).toHaveBeenCalledTimes(1);
   });
+
+  /** A cancelled compaction's partial row is built here, not by sendCompletion,
+   *  so it carries no marker unless the disconnect path stamps one: without it
+   *  the row reads as an answer to the message it hangs off and keeps that
+   *  message's rerun controls. */
+  it('stamps a partial response saved on disconnect with the compaction identity', async () => {
+    await firePartialDisconnect(
+      { id: 'user-123' },
+      { createdAt: 1000 },
+      {
+        body: { compact: true },
+        aggregatedContent: [
+          {
+            type: 'summary',
+            content: [{ type: 'text', text: 'Half a summary' }],
+            summarizing: true,
+          },
+        ],
+      },
+    );
+
+    const [, savedMessage] = mockSaveMessage.mock.calls[0];
+    expect(savedMessage).toMatchObject({
+      messageId: 'response-message',
+      unfinished: true,
+      error: false,
+      content: [{ type: 'summary', summarizing: true, initiatedBy: 'user' }],
+    });
+  });
+
+  /** The disconnect save runs while the generation is still live and the
+   *  completing run overwrites the row, so it must not report a failure that
+   *  has not happened: no typed failure is invented for a compaction whose
+   *  snapshot carries no summary or error part. */
+  it('saves a non-outcome compaction partial on disconnect without a synthesized failure', async () => {
+    await firePartialDisconnect(
+      { id: 'user-123' },
+      { createdAt: 1000 },
+      {
+        body: { compact: true },
+        aggregatedContent: [{ type: 'think', think: 'Picking what to summarize' }],
+      },
+    );
+
+    const [, savedMessage] = mockSaveMessage.mock.calls[0];
+    expect(savedMessage).toMatchObject({
+      unfinished: true,
+      error: false,
+      content: [{ type: 'think', think: 'Picking what to summarize' }],
+    });
+    expect(savedMessage.content).toHaveLength(1);
+  });
+  /** The settling path (completion, error, abort) owns the compaction's final
+   *  row: a disconnect snapshot landing after it would reopen the settled
+   *  turn as an unfinished response. */
+  it('skips the compaction partial save when the job record has settled', async () => {
+    await firePartialDisconnect(
+      { id: 'user-123' },
+      { createdAt: 1000, status: 'error' },
+      {
+        body: { compact: true },
+        aggregatedContent: [{ type: 'text', text: 'Partial response' }],
+      },
+    );
+
+    expect(mockSaveMessage).not.toHaveBeenCalled();
+  });
+
+  /** Ordinary turns keep the pre-change behavior exactly: their snapshot is
+   *  the fallback row even when the job record has settled, because the
+   *  terminal row write may still fail. */
+  it('still persists an ordinary 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).toHaveBeenCalledTimes(1);
+    const [, savedMessage] = mockSaveMessage.mock.calls[0];
+    expect(savedMessage).toMatchObject({ unfinished: true, error: false });
+  });
+  /** The terminal write failed and settled for a reconciliation frame: the
+   *  snapshot is the turn's only row, so it persists with the terminal
+   *  outcome (the typed failure for content with no summary) and envelope. */
+  it('promotes a reconciled compaction snapshot to the terminal row', async () => {
+    mockGetMessages.mockResolvedValue([{ _id: 'existing-anchor' }]);
+    await firePartialDisconnect(
+      { id: 'user-123' },
+      {
+        createdAt: 1000,
+        status: 'error',
+        finalEvent: JSON.stringify({ final: true, reconcile: true }),
+      },
+      {
+        body: { compact: true },
+        aggregatedContent: [{ type: 'think', think: 'Picking what to summarize' }],
+      },
+    );
+
+    expect(mockSaveMessage).toHaveBeenCalledTimes(1);
+    const [, savedMessage] = mockSaveMessage.mock.calls[0];
+    expect(savedMessage).toMatchObject({ unfinished: false, error: true });
+    expect(savedMessage.content).toEqual([
+      { type: 'think', think: 'Picking what to summarize' },
+      expect.objectContaining({ type: 'error', initiatedBy: 'user' }),
+    ]);
+  });
+  /** Without a persisted anchor the promotion is withheld: the fallback must
+   *  not recreate the orphan the absent-anchor abort suppressed. */
+  it('withholds a reconciled snapshot whose anchor was never persisted', async () => {
+    await firePartialDisconnect(
+      { id: 'user-123' },
+      {
+        createdAt: 1000,
+        status: 'error',
+        finalEvent: JSON.stringify({ final: true, reconcile: true }),
+      },
+      {
+        body: { compact: true },
+        aggregatedContent: [{ type: 'text', text: 'Partial response' }],
+      },
+    );
+
+    expect(mockSaveMessage).not.toHaveBeenCalled();
+  });
+
+  /** An empty snapshot on a reconciled compaction still records the turn:
+   *  the terminal synthesis appends the typed failure from nothing. */
+  it('synthesizes the terminal outcome for an empty reconciled compaction snapshot', async () => {
+    mockGetMessages.mockResolvedValue([{ _id: 'existing-anchor' }]);
+    await firePartialDisconnect(
+      { id: 'user-123' },
+      {
+        createdAt: 1000,
+        status: 'error',
+        finalEvent: JSON.stringify({ final: true, reconcile: true }),
+      },
+      { body: { compact: true }, aggregatedContent: [] },
+    );
+
+    expect(mockSaveMessage).toHaveBeenCalledTimes(1);
+    const [, savedMessage] = mockSaveMessage.mock.calls[0];
+    expect(savedMessage).toMatchObject({ unfinished: false, error: true });
+    expect(savedMessage.content).toEqual([
+      expect.objectContaining({ type: 'error', initiatedBy: 'user' }),
+    ]);
+  });
+
+  /** A completed claim's promoted snapshot settles as a finished row, not an
+   *  error. */
+  it('settles a completed reconciled snapshot as a finished row', async () => {
+    mockGetMessages.mockResolvedValue([{ _id: 'existing-anchor' }]);
+    await firePartialDisconnect(
+      { id: 'user-123' },
+      {
+        createdAt: 1000,
+        status: 'complete',
+        finalEvent: JSON.stringify({ final: true, reconcile: true }),
+      },
+      {
+        body: { compact: true },
+        aggregatedContent: [
+          {
+            type: 'summary',
+            content: [{ type: 'text', text: 'A finished checkpoint.' }],
+            boundary: { messageId: 'm', contentIndex: 0 },
+          },
+        ],
+      },
+    );
+
+    expect(mockSaveMessage).toHaveBeenCalledTimes(1);
+    const [, savedMessage] = mockSaveMessage.mock.calls[0];
+    expect(savedMessage).toMatchObject({ unfinished: false, error: false });
+  });
 });

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

@@ -257,6 +257,15 @@ jest.mock('@librechat/api', () => ({
   ).getSteerRecoveryFailure,
   getAgentErrorMetadata: (...args) =>
     jest.requireActual('@librechat/api').getAgentErrorMetadata(...args),
+  markAbortedCompactionContent: (...args) =>
+    jest.requireActual('@librechat/api').markAbortedCompactionContent(...args),
+  isSettledJobRecord: (...args) => jest.requireActual('@librechat/api').isSettledJobRecord(...args),
+  resolveDisconnectSnapshotMode: (...args) =>
+    jest.requireActual('@librechat/api').resolveDisconnectSnapshotMode(...args),
+  resolveReconciledSnapshotEnvelope: (...args) =>
+    jest.requireActual('@librechat/api').resolveReconciledSnapshotEnvelope(...args),
+  settleExistingRowsBeforeErrorTurn: (...args) =>
+    jest.requireActual('@librechat/api').settleExistingRowsBeforeErrorTurn(...args),
   sendEvent: jest.fn(),
   logAgentMemorySnapshot: jest.fn(),
   isScheduleFireRequest: (...args) => mockIsScheduleFireRequest(...args),

api/server/controllers/agents/request.js

@@ -52,6 +52,10 @@ const {
   resolvePersistableCodeEnvironmentDecision,
   getFailedTurnTraceFields,
   resolveFailedTurnContent,
+  settleExistingRowsBeforeErrorTurn,
+  resolveDisconnectSnapshotMode,
+  resolveReconciledSnapshotEnvelope,
+  markAbortedCompactionContent,
 } = require('@librechat/api');
 const { disposeClient } = require('~/server/cleanup');
 const {
@@ -426,23 +430,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:
@@ -453,6 +440,29 @@ 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 settlement = 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',
+        }),
+    });
+    if (settlement.covered) {
+      return;
+    }
+    /** The anchor-shaped collision redirects the error row to the failed
+     *  run's own response id, so it can never overwrite the anchor. */
+    if (settlement.errorRowMessageId != null) {
+      errorMessageId = settlement.errorRowMessageId;
+    }
+
     const context = 'api/server/controllers/agents/request.js - failed turn';
     const endpoint = endpointOption?.endpoint;
     const model = getAgentResponseModel(req, endpointOption);
@@ -1663,6 +1673,9 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
          *  no user message of its own, the response parented onto an
          *  existing message. A reconnecting client rebuilds it that way. */
         ...((isRegenerate || isCompaction) && { isRegenerate: true }),
+        /** The job record is where the abort paths learn the turn was a
+         *  compaction: they run after the request that created the job. */
+        ...(isCompaction && { compact: true }),
         ...(scheduleId
           ? {
               scheduleId,
@@ -1849,13 +1862,10 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
      * overwrite this with the complete response using the same messageId pattern.
      */
     job.emitter.on('allSubscribersLeft', async (aggregatedContent) => {
-      if (partialResponseSaved || !aggregatedContent || aggregatedContent.length === 0) {
-        return;
-      }
-
-      const persistableContent = filterPersistableAbortContent(aggregatedContent);
-      if (persistableContent.length === 0) {
-        logger.debug('[ResumableAgentController] No persistable content to save partial response');
+      /** Empty content is rejected only after the snapshot mode is known: a
+       *  reconciled compaction synthesizes its terminal outcome from nothing,
+       *  while a live run has nothing persistable. */
+      if (partialResponseSaved || !aggregatedContent) {
         return;
       }
 
@@ -1868,6 +1878,47 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
         return;
       }
 
+      /** How this snapshot may persist is decided in @librechat/api: live
+       *  runs keep the marker-only shape, a failed terminal write (settled
+       *  for a reconciliation frame) promotes the snapshot to the turn's
+       *  terminal row, and a durably settled compaction withholds it. */
+      const snapshotMode = await resolveDisconnectSnapshotMode(
+        isCompaction,
+        jobRecord,
+        jobCreatedAt,
+        {
+          anchorExists: async () =>
+            (
+              await getMessages(
+                {
+                  user: userId,
+                  messageId: resumeState.userMessage.messageId,
+                  conversationId,
+                },
+                '_id',
+              )
+            ).length > 0,
+        },
+      );
+      if (snapshotMode === 'skip') {
+        logger.debug(
+          '[ResumableAgentController] Skipping compaction partial save for a settled job',
+        );
+        return;
+      }
+      if (snapshotMode !== 'terminal' && aggregatedContent.length === 0) {
+        return;
+      }
+      const persistableContent = markAbortedCompactionContent(
+        filterPersistableAbortContent(aggregatedContent),
+        isCompaction,
+        { synthesizeFailure: snapshotMode === 'terminal' },
+      );
+      if (persistableContent.length === 0) {
+        logger.debug('[ResumableAgentController] No persistable content to save partial response');
+        return;
+      }
+
       partialResponseSaved = true;
       const responseConversationId = resumeState.conversationId || conversationId;
       /** The run publishes its calibration and fading tiers onto the job; a
@@ -1885,8 +1936,12 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
           parentMessageId: resumeState.userMessage.messageId,
           sender: client?.sender ?? 'AI',
           content: persistableContent,
-          unfinished: true,
-          error: false,
+          /** A snapshot promoted to the terminal row settles with the
+           *  envelope its reconciled claim's status dictates; a live-run
+           *  snapshot keeps the live shape. */
+          ...(snapshotMode === 'terminal'
+            ? resolveReconciledSnapshotEnvelope(jobRecord?.status)
+            : { unfinished: true, error: false }),
           isCreatedByUser: false,
           user: userId,
           endpoint: endpointOption.endpoint,