fix(chat): 修复长回合终态历史落库失败导致对话丢失

crates/agent-gui/src/lib/chat/history/chatHistory.ts

@@ -429,6 +429,11 @@ function buildChatHistoryConversationInput(params: {
     state,
   } = params;
 
+  // Header totals must stay anchored on state.meta. After a conversation is
+  // reopened from history, state.segments only holds the active segment while
+  // meta.totalMessageCount still counts every sealed row in SQLite — summing
+  // the in-memory segments would undercount and trip the backend segment-sum
+  // consistency check on every persist.
   return {
     id: conversationId,
     title,
@@ -446,13 +451,23 @@ function buildChatHistoryConversationInput(params: {
   };
 }
 
+function resolveSegmentMessageCount(segment: StoredContextSegment): number {
+  // Live messages are the source of truth. Fall back to messageCount only when
+  // the messages array is absent (defensive); an empty array means zero messages.
+  if (Array.isArray(segment.messages)) {
+    return segment.messages.length;
+  }
+  return typeof segment.messageCount === "number" ? segment.messageCount : 0;
+}
+
 function buildChatHistorySegmentInput(segment: StoredContextSegment): ChatHistorySegmentWireRecord {
+  const messageCount = resolveSegmentMessageCount(segment);
   return {
     segmentIndex: segment.segmentIndex,
     segmentId: segment.segmentId,
     summaryJson: segment.summary ? JSON.stringify(segment.summary) : undefined,
-    messagesJson: JSON.stringify(segment.messages),
-    messageCount: segment.messageCount,
+    messagesJson: JSON.stringify(segment.messages ?? []),
+    messageCount,
     startMessageId: segment.startMessageId,
     endMessageId: segment.endMessageId,
     createdAt: segment.createdAt,
@@ -542,10 +557,58 @@ type PersistConversationRuntimeParams = {
   commitPersistenceCursor: (cursor: ConversationPersistenceCursor) => void;
 };
 
+function findSegmentByIndex(
+  state: ConversationViewState,
+  segmentIndex: number,
+): StoredContextSegment | undefined {
+  return state.segments.find((segment) => segment.segmentIndex === segmentIndex);
+}
+
+function conversationInputForCursor(
+  conversation: ChatHistoryConversationInput,
+  state: ConversationViewState,
+  activeSegmentIndex: number,
+): ChatHistoryConversationInput {
+  // Each intermediate append must advertise exactly the totals that exist
+  // after that step lands: the backend append precondition requires
+  // totalSegmentCount == stored count + 1, and the consistency check compares
+  // the header against COUNT/SUM over all rows inside the same transaction.
+  //
+  // Subtract the not-yet-appended in-memory segments from the final header
+  // total instead of re-summing in-memory segments from zero: after a
+  // conversation is reopened from history, the sealed rows before the loaded
+  // active segment exist only in SQLite.
+  const pendingBeyondStep = state.segments
+    .filter((segment) => segment.segmentIndex > activeSegmentIndex)
+    .reduce((sum, segment) => sum + resolveSegmentMessageCount(segment), 0);
+  const totalSegmentCount = activeSegmentIndex + 1;
+  const totalMessageCount = Math.max(0, conversation.totalMessageCount - pendingBeyondStep);
+  let contextMetaJson = conversation.contextMetaJson;
+  try {
+    const meta = JSON.parse(conversation.contextMetaJson) as Record<string, unknown>;
+    contextMetaJson = JSON.stringify({
+      ...meta,
+      activeSegmentIndex,
+      totalSegmentCount,
+      totalMessageCount,
+    });
+  } catch {
+    // Keep the original payload if meta is not JSON; header counts still win.
+  }
+  return {
+    ...conversation,
+    contextMetaJson,
+    activeSegmentIndex,
+    totalSegmentCount,
+    totalMessageCount,
+  };
+}
+
 async function writeConversationRuntime(
   conversation: ChatHistoryConversationInput,
   cursor: ConversationPersistenceCursor | null,
   state: ConversationViewState,
+  commitPersistenceCursor: (cursor: ConversationPersistenceCursor) => void,
 ) {
   const activeSegment = getActiveSegment(state);
   if (!activeSegment) {
@@ -556,60 +619,91 @@ async function writeConversationRuntime(
     if (state.segments[0]?.segmentIndex !== 0) {
       throw new Error("已存在的历史会话缺少持久化游标");
     }
-    return upsertChatHistoryRaw({
+    const summary = await upsertChatHistoryRaw({
       ...conversation,
       segments: state.segments.map(buildChatHistorySegmentInput),
     });
+    commitPersistenceCursor({
+      activeSegmentIndex: activeSegment.segmentIndex,
+      activeSegmentId: activeSegment.segmentId,
+    });
+    return summary;
+  }
+
+  if (activeSegment.segmentIndex < cursor.activeSegmentIndex) {
+    throw new Error(
+      `不支持的历史分段回退：${cursor.activeSegmentIndex} -> ${activeSegment.segmentIndex}`,
+    );
   }
 
   if (activeSegment.segmentIndex === cursor.activeSegmentIndex) {
     if (activeSegment.segmentId !== cursor.activeSegmentId) {
       throw new Error("活跃历史分段身份与持久化游标不一致");
     }
-    return upsertChatHistoryActiveSegmentRaw({
+    const summary = await upsertChatHistoryActiveSegmentRaw({
       conversation,
       segment: buildChatHistorySegmentInput(activeSegment),
     });
+    commitPersistenceCursor({
+      activeSegmentIndex: activeSegment.segmentIndex,
+      activeSegmentId: activeSegment.segmentId,
+    });
+    return summary;
   }
 
-  if (activeSegment.segmentIndex === cursor.activeSegmentIndex + 1) {
-    const previousSegment = state.segments.find(
-      (segment) => segment.segmentIndex === cursor.activeSegmentIndex,
-    );
+  // Catch up one segment at a time when the in-memory active segment jumped
+  // ahead of the durable cursor (e.g. multiple compactions between persists).
+  // Previously this threw "不支持的历史分段跳变" and left the DB on the
+  // user-only snapshot after a long agent turn.
+  let workingCursor: ConversationPersistenceCursor = { ...cursor };
+  let summary: ChatHistorySummary | null = null;
+
+  while (workingCursor.activeSegmentIndex < activeSegment.segmentIndex) {
+    const previousSegment = findSegmentByIndex(state, workingCursor.activeSegmentIndex);
+    const nextSegment = findSegmentByIndex(state, workingCursor.activeSegmentIndex + 1);
     if (!previousSegment) {
       throw new Error("追加历史分段时缺少待封存的上一活跃分段");
     }
-    if (previousSegment.segmentId !== cursor.activeSegmentId) {
+    if (!nextSegment) {
+      throw new Error(`追加历史分段时缺少目标分段：${workingCursor.activeSegmentIndex + 1}`);
+    }
+    if (previousSegment.segmentId !== workingCursor.activeSegmentId) {
       throw new Error("待封存历史分段身份与持久化游标不一致");
     }
-    return appendChatHistorySegmentRaw({
-      conversation,
+
+    summary = await appendChatHistorySegmentRaw({
+      conversation: conversationInputForCursor(conversation, state, nextSegment.segmentIndex),
       previousSegment: buildChatHistorySegmentInput(previousSegment),
-      segment: buildChatHistorySegmentInput(activeSegment),
+      segment: buildChatHistorySegmentInput(nextSegment),
     });
+    workingCursor = {
+      activeSegmentIndex: nextSegment.segmentIndex,
+      activeSegmentId: nextSegment.segmentId,
+    };
+    // Commit after every successful append so a later failure can resume from
+    // the durable frontier instead of replaying a sealed segment.
+    commitPersistenceCursor(workingCursor);
   }
 
-  throw new Error(
-    `不支持的历史分段跳变：${cursor.activeSegmentIndex} -> ${activeSegment.segmentIndex}`,
-  );
+  if (activeSegment.segmentId !== workingCursor.activeSegmentId) {
+    throw new Error("活跃历史分段身份与持久化游标不一致");
+  }
+  if (!summary) {
+    throw new Error(
+      `不支持的历史分段跳变：${cursor.activeSegmentIndex} -> ${activeSegment.segmentIndex}`,
+    );
+  }
+  return summary;
 }
 
 export async function persistConversationRuntime(params: PersistConversationRuntimeParams) {
   return withConversationWriteLock(params.conversationId, async () => {
     const conversation = buildChatHistoryConversationInput(params);
-    const summary = await writeConversationRuntime(
+    return writeConversationRuntime(
       conversation,
       params.getPersistenceCursor(),
       params.state,
+      params.commitPersistenceCursor,
     );
-    const activeSegment = getActiveSegment(params.state);
-    if (!activeSegment) {
-      throw new Error("持久化成功后缺少活跃分段");
-    }
-    params.commitPersistenceCursor({
-      activeSegmentIndex: activeSegment.segmentIndex,
-      activeSegmentId: activeSegment.segmentId,
-    });
-    return summary;
   });
 }

crates/agent-gui/src/pages/chat/runtime/chatRunFinalization.ts

@@ -1,5 +1,9 @@
 export const CHAT_RUN_FINALIZATION_TIMEOUT_MS = 2_000;
 
+/** Terminal history writes get a few short retries before the run is marked failed. */
+export const TERMINAL_HISTORY_PERSIST_MAX_ATTEMPTS = 3;
+export const TERMINAL_HISTORY_PERSIST_RETRY_DELAY_MS = 150;
+
 export function releaseChatRunUi(params: {
   clearAbortController: () => void;
   clearSendingState: () => void;
@@ -10,6 +14,72 @@ export function releaseChatRunUi(params: {
   params.clearToolStatus();
 }
 
+function delay(ms: number): Promise<void> {
+  return new Promise((resolve) => {
+    setTimeout(resolve, Math.max(0, ms));
+  });
+}
+
+/**
+ * Retry a terminal history write a few times. Transient SQLite / IPC failures
+ * during the final persist otherwise leave the DB stuck on the user-only
+ * snapshot while memory extraction still runs from in-memory state.
+ */
+export async function persistTerminalHistoryWithRetry(
+  persist: () => Promise<boolean>,
+  options?: {
+    maxAttempts?: number;
+    retryDelayMs?: number;
+    sleep?: (ms: number) => Promise<void>;
+    onRetry?: (error: unknown, attempt: number, maxAttempts: number) => void;
+  },
+): Promise<boolean> {
+  const maxAttempts = Math.max(1, options?.maxAttempts ?? TERMINAL_HISTORY_PERSIST_MAX_ATTEMPTS);
+  const retryDelayMs = Math.max(
+    0,
+    options?.retryDelayMs ?? TERMINAL_HISTORY_PERSIST_RETRY_DELAY_MS,
+  );
+  const sleep = options?.sleep ?? delay;
+  let lastError: unknown = null;
+  // null = no terminal outcome yet; true = last attempt returned false; false = threw
+  let lastAttemptWasSoftFalse: boolean | null = null;
+
+  for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
+    try {
+      const persisted = await persist();
+      if (persisted) {
+        return true;
+      }
+      lastAttemptWasSoftFalse = true;
+      lastError = new Error("history persist returned false");
+    } catch (error) {
+      lastAttemptWasSoftFalse = false;
+      lastError = error;
+    }
+
+    if (attempt >= maxAttempts) {
+      break;
+    }
+    options?.onRetry?.(lastError, attempt, maxAttempts);
+    if (retryDelayMs > 0) {
+      await sleep(retryDelayMs * attempt);
+    }
+  }
+
+  // Preserve the historical contract: a soft `false` from persist stays a
+  // boolean failure, while thrown errors remain exceptional.
+  if (lastAttemptWasSoftFalse) {
+    return false;
+  }
+  if (lastError instanceof Error) {
+    throw lastError;
+  }
+  if (lastError != null) {
+    throw new Error(String(lastError));
+  }
+  return false;
+}
+
 export async function settleChatRunFinalization(
   finalization: Promise<unknown>,
   timeoutMs = CHAT_RUN_FINALIZATION_TIMEOUT_MS,
@@ -35,9 +105,15 @@ export async function settleChatRunFinalization(
 export async function trackTerminalHistoryPersist(
   persist: () => Promise<boolean>,
   markFailed: () => void,
+  options?: {
+    maxAttempts?: number;
+    retryDelayMs?: number;
+    sleep?: (ms: number) => Promise<void>;
+    onRetry?: (error: unknown, attempt: number, maxAttempts: number) => void;
+  },
 ): Promise<boolean> {
   try {
-    const persisted = await persist();
+    const persisted = await persistTerminalHistoryWithRetry(persist, options);
     if (!persisted) {
       markFailed();
     }

crates/agent-gui/src/pages/chat/turns/runAgentConversationTurn.ts

@@ -1111,7 +1111,37 @@ export async function runAgentConversationTurn(params: RunAgentConversationTurnP
     });
   };
 
-  if (showSilentMemoryExtraction && shouldRunMemoryExtraction) {
+  const persistCompletedState = (state: ConversationViewState) =>
+    persistConversationWithHistorySync({
+      conversationId,
+      sessionId,
+      providerId,
+      model,
+      cwd: conversationCwd,
+      state,
+      fallbackTitle,
+      createdAt,
+      titlePromise,
+    });
+
+  const pendingTerminalAssistantMeta = pendingTerminalAssistantMetaRef.current;
+  if (pendingTerminalAssistantMeta) {
+    commitAssistantRoundMeta(
+      pendingTerminalAssistantMeta.assistant,
+      pendingTerminalAssistantMeta.round,
+    );
+  }
+  hookLifecycle.endAgent();
+
+  applyConversationState(finalState);
+  freezeGatewayFinalProjection(finalState, true);
+  settleLiveTranscript(transcriptStore);
+  const historyPersisted = await persistCompletedState(finalState);
+
+  // Memory extraction reads the in-memory final state. Only run it after the
+  // durable history write succeeds so we never keep "memory has the answer,
+  // chat history only has the user prompt" after a failed final persist.
+  if (historyPersisted && showSilentMemoryExtraction && shouldRunMemoryExtraction) {
     const extraction = await runPostTurnMemoryExtraction({
       roundOffset: memoryRoundOffset,
       onTurnStart: (round) => {
@@ -1241,29 +1271,13 @@ export async function runAgentConversationTurn(params: RunAgentConversationTurnP
       );
     }
   }
-  const pendingTerminalAssistantMeta = pendingTerminalAssistantMetaRef.current;
-  if (pendingTerminalAssistantMeta) {
-    commitAssistantRoundMeta(
-      pendingTerminalAssistantMeta.assistant,
-      pendingTerminalAssistantMeta.round,
-    );
+  if (completedState !== finalState) {
+    applyConversationState(completedState);
+    freezeGatewayFinalProjection(completedState, true);
+    settleLiveTranscript(transcriptStore);
+    await persistCompletedState(completedState);
   }
-  hookLifecycle.endAgent();
-  applyConversationState(completedState);
-  freezeGatewayFinalProjection(completedState, true);
-  settleLiveTranscript(transcriptStore);
-  await persistConversationWithHistorySync({
-    conversationId,
-    sessionId,
-    providerId,
-    model,
-    cwd: conversationCwd,
-    state: completedState,
-    fallbackTitle,
-    createdAt,
-    titlePromise,
-  });
-  if (!showSilentMemoryExtraction && shouldRunMemoryExtraction) {
+  if (historyPersisted && !showSilentMemoryExtraction && shouldRunMemoryExtraction) {
     void runPostTurnMemoryExtraction();
   }
 }