fix(messaging): implement hydration pipeline to prevent history loss on page entry

This addresses the issue where entering a chat page would only display recent
websocket messages, swallowing historical messages.

The fix implements a proper hydration pipeline in MessageSyncService that:
1. Tracks hydrated state per conversation via `hydratedConversations` Set
2. Unhydrated conversations run full pipeline: local contiguous range check,
   incremental sync above maxSeq, gap fill below minSeq, mark hydrated
3. Hydrated conversations only do lightweight incremental sync using trusted baseline
4. Clear conversation now also clears in-memory state and hydrated flag

Also adds:
- `getContiguousRange()` method to find local message continuity gaps
- `MESSAGES_PAGE_SIZE` (20) and `MAX_GAP_FILL_ROUNDS` (5) constants
- Test mocks with configurable state for hydration scenarios
- `messageHydration.test.ts` unit tests

src/database/repositories/MessageRepository.ts

@@ -9,6 +9,16 @@ export interface IMessageRepository {
   getByConversation(conversationId: string, limit?: number): Promise<CachedMessage[]>;
   getMaxSeq(conversationId: string): Promise<number>;
   getMinSeq(conversationId: string): Promise<number>;
+  /**
+   * 返回本地某会话从最大 seq 往下的「连续区间」。
+   * - maxSeq: 该会话本地最大的 seq（0 表示无消息）
+   * - minSeq: 从 maxSeq 往下连续递减遇到第一个缺口时的下沿（即连续段的最早 seq）
+   *
+   * 用于 hydration 管线判断「最新端往前是否存在历史缺口」：
+   * 若 minSeq > 1，说明 [minSeq-1, ...] 更早的消息本地缺失，需要回填。
+   * 应用层遍历而非 SQL 窗口函数，保证跨 SQLite 版本兼容、易测试。
+   */
+  getContiguousRange(conversationId: string): Promise<{ maxSeq: number; minSeq: number }>;
   getBeforeSeq(conversationId: string, beforeSeq: number, limit?: number): Promise<CachedMessage[]>;
   getCount(conversationId: string): Promise<number>;
   getCountBeforeSeq(conversationId: string, beforeSeq: number): Promise<number>;
@@ -79,6 +89,27 @@ export class MessageRepository implements IMessageRepository {
     return r?.minSeq || 0;
   }
 
+  async getContiguousRange(conversationId: string): Promise<{ maxSeq: number; minSeq: number }> {
+    // 单次查询拉出该会话所有有效 seq，按 DESC 排序；应用层从最大值向下找第一个缺口。
+    // 会话消息量通常可控，且仅对未 hydrated 的会话执行一次，开销可接受。
+    const rows = await this.dataSource.query<{ seq: number }>(
+      `SELECT seq FROM messages WHERE conversationId = ? AND seq > 0 ORDER BY seq DESC`,
+      [conversationId]
+    );
+    if (rows.length === 0) return { maxSeq: 0, minSeq: 0 };
+
+    let maxSeq = rows[0].seq;
+    let minSeq = maxSeq;
+    for (let i = 1; i < rows.length; i++) {
+      if (rows[i].seq === minSeq - 1) {
+        minSeq = rows[i].seq; // 连续，向下扩展
+      } else {
+        break; // 遇到缺口，停止
+      }
+    }
+    return { maxSeq, minSeq };
+  }
+
   async getBeforeSeq(conversationId: string, beforeSeq: number, limit: number = 20): Promise<CachedMessage[]> {
     const rows = await this.dataSource.query<any>(
       `SELECT * FROM messages WHERE conversationId = ? AND seq < ? ORDER BY seq DESC LIMIT ?`,

src/screens/message/PrivateChatInfoScreen.tsx

@@ -133,6 +133,9 @@ const PrivateChatInfoScreen: React.FC = () => {
           onPress: async () => {
             try {
               await messageRepository.clearConversation(conversationId);
+              // 同步清空内存消息 + hydrated 标志，保证返回聊天页时状态与 DB 一致。
+              // 否则 hydrated 残留 + 内存旧消息会显示已清空的记录。
+              useMessageStore.getState().clearMessages(conversationId);
               Alert.alert('成功', '聊天记录已清空');
             } catch (error) {
               Alert.alert('错误', '清空聊天记录失败');

src/screens/message/components/ChatScreen/useChatScreen.ts

@@ -1515,7 +1515,10 @@ export const useChatScreen = (props?: ChatScreenProps) => {
     try {
       setHasMoreHistory(true);
       await messageRepository.clearConversation(conversationId);
-      // 刷新消息列表
+      // 同步清空内存消息 + hydrated 标志，保证内存/DB/hydrated 三者一致。
+      // 否则 hydrated 残留会让 refreshMessages 走轻量增量同步，清空后仍显示旧消息。
+      useMessageStore.getState().clearMessages(conversationId);
+      // 刷新消息列表（hydrated 已清，会重新走完整 hydration 管线）
       await refreshMessages();
     } catch (error) {
       console.error('清空会话失败:', error);

src/stores/message/__tests__/messageHydration.test.ts

@@ -0,0 +1,323 @@
+/**
+ * 消息 hydration 管线单测（A+B+C 融合方案）
+ *
+ * 锁定的不变量（直接对应「从消息栏进入聊天页只显示一条消息、历史被吞」的回归）：
+ *
+ * 1. 【方案 A 核心回归】内存已有 WS 零散最新消息时，进入会话仍能从本地 DB 加载历史，
+ *    不再因为「内存非空」而跳过本地读取、误判 baseline。
+ * 2. 【方案 B 连续回填】本地连续区间存在下缺口时，管线循环 loadMore 回填直到连续到顶。
+ * 3. 【方案 B 上限保护】历史巨大时回填在 MAX_GAP_FILL_ROUNDS 后停止，不卡死。
+ * 4. 【方案 C 门控】首次进入走完整管线并标记 hydrated；重入只做轻量增量同步。
+ * 5. 回填后内存消息按 seq 严格升序、无重复、无丢失。
+ *
+ * mock 机制：jest moduleNameMapper 自动把 @/services/message、@/database 重定向到
+ * mocks/messageServiceMock、mocks/databaseMock。测试通过 __setMessagesHandler /
+ * __setMessagesByConversation 配置响应，beforeEach 调 __reset 复原。
+ */
+
+import { useMessageStore } from '../store';
+import { MessageSyncService } from '../services/MessageSyncService';
+import { messageService } from '@/services/message';
+import { messageRepository } from '@/database';
+import type { MessageResponse } from '../../../types/dto';
+import { MAX_GAP_FILL_ROUNDS, MESSAGES_PAGE_SIZE } from '../constants';
+
+// mock 注入的测试辅助方法（运行时由 mocks/databaseMock、mocks/messageServiceMock 提供，
+// 类型上不存在，统一用 any 别名调用，与现有 (xxx as any).reset() 风格一致）
+const repoMock = messageRepository as any;
+const serviceMock = messageService as any;
+
+// ---- 测试辅助 ----
+
+function makeMessage(id: string, seq: number, extra: Partial<MessageResponse> = {}): MessageResponse {
+  return {
+    id,
+    conversation_id: 'c1',
+    sender_id: 'u1',
+    seq,
+    segments: [{ type: 'text', data: { text: `msg-${seq}` } }],
+    created_at: `2024-01-01T00:00:${String(seq).padStart(2, '0')}Z`,
+    status: 'normal',
+    ...extra,
+  };
+}
+
+/** 生成 [from..to] 区间的消息（含端点），seq 升序。 */
+function makeRange(from: number, to: number, prefix = 'm'): MessageResponse[] {
+  const out: MessageResponse[] = [];
+  for (let s = from; s <= to; s++) {
+    out.push(makeMessage(`${prefix}-${s}`, s));
+  }
+  return out;
+}
+
+/**
+ * 模拟真实后端的 getMessages 行为：
+ * - afterSeq 指定：返回 seq > afterSeq 的「新消息」（默认空，可由 extraOverrides 注入）
+ * - beforeSeq 指定：返回 [beforeSeq-limit, beforeSeq-1] 区间的历史（分页回填）
+ * - 都未指定：返回最近 limit 条快照（默认空，可由 snapshotOverride 注入）
+ *
+ * 这样回填循环每轮 beforeSeq 请求都能拿到真实历史，模拟线上行为。
+ */
+function makeServerHandler(opts: {
+  topSeq?: number; // 会话历史的最大 seq（决定回填到顶）；不传则按 beforeSeq 递减
+  extraAbove?: MessageResponse[]; // afterSeq 增量返回的新消息
+  snapshot?: MessageResponse[]; // 快照响应覆盖
+} = {}): (args: any) => any {
+  return ({ afterSeq, beforeSeq, limit }) => {
+    const pageSize = limit || MESSAGES_PAGE_SIZE;
+    if (afterSeq !== undefined) {
+      return { messages: opts.extraAbove || [], total: 0, page: 1, page_size: pageSize };
+    }
+    if (beforeSeq !== undefined) {
+      const from = Math.max(1, beforeSeq - pageSize);
+      const to = beforeSeq - 1;
+      if (to < 1) return { messages: [], total: 0, page: 1, page_size: pageSize };
+      return { messages: makeRange(from, to), total: to, page: 1, page_size: pageSize };
+    }
+    // 快照
+    return { messages: opts.snapshot || [], total: 0, page: 1, page_size: pageSize };
+  };
+}
+
+/** 构造 MessageSyncService，依赖全部用最小存根。 */
+function createSyncService(): MessageSyncService {
+  return new MessageSyncService(
+    () => 'me',
+    { markAsRead: async () => {}, markAllAsRead: async () => {} } as any,
+    {
+      getSenderInfo: async () => null,
+      enrichMessagesWithSenderInfo: () => {},
+    } as any,
+    undefined
+  );
+}
+
+describe('消息 hydration 管线', () => {
+  beforeEach(() => {
+    useMessageStore.getState().reset();
+    repoMock.__reset();
+    serviceMock.__reset();
+  });
+
+  describe('方案 A：baseline 修正（核心回归）', () => {
+    it('内存已有 WS 零散最新消息时，仍从本地 DB 加载更早的历史', async () => {
+      // 场景：WS 已把 seq=49 写入内存（模拟用户停留在消息栏时收到的最新推送），
+      //       本地 DB 存有 seq=1~48 的完整历史。
+      // 修复前：existingMessagesAtStart.length>0 → 跳过本地读取 → after_seq=49 → 历史全丢。
+      // 修复后：无条件读本地补全 → 内存得到 seq=1~49。
+      useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('ws-49', 49));
+      repoMock.__setMessagesByConversation('c1', makeRange(1, 48));
+
+      // 后端：after_seq=49 无新消息；beforeSeq 请求返回分页历史（模拟真实后端）
+      serviceMock.__setMessagesHandler(makeServerHandler());
+
+      const sync = createSyncService();
+      await sync.fetchMessages('c1');
+
+      const msgs = useMessageStore.getState().getMessages('c1');
+      // 关键断言：应能看到完整历史，而非只有 seq=49 一条
+      expect(msgs).toHaveLength(49);
+      expect(msgs.map((m) => m.seq)).toEqual(expect.arrayContaining(makeRange(1, 49).map((m) => m.seq)));
+      expect(msgs[0].seq).toBe(1);
+      expect(msgs[msgs.length - 1].seq).toBe(49);
+      // 已 hydrated
+      expect(useMessageStore.getState().isHydrated('c1')).toBe(true);
+    });
+
+    it('内存与本地均空时走冷启动快照分支', async () => {
+      // 冷启动：本地无数据，拉取最近 50 条快照
+      const snapshot = makeRange(1, 50);
+      serviceMock.__setMessagesHandler(makeServerHandler({ snapshot }));
+
+      const sync = createSyncService();
+      await sync.fetchMessages('c1');
+
+      const msgs = useMessageStore.getState().getMessages('c1');
+      expect(msgs).toHaveLength(50);
+      expect(useMessageStore.getState().isHydrated('c1')).toBe(true);
+    });
+  });
+
+  describe('方案 B：连续区间回填', () => {
+    it('本地连续区间有下缺口时，循环 loadMore 回填到顶', async () => {
+      // 场景：本地 DB 只有 seq=45~49（minSeq=45 > 1，存在下缺口），
+      //       服务端按 beforeSeq 分段返回历史。
+      // 注意：getByConversation 默认读 MESSAGES_PAGE_SIZE=20 条，本地只有 5 条，
+      //       故 Step 1 先把 5 条并入内存；Step 2 上缺口无新消息；
+      //       Step 3 回填：beforeSeq=45 → 服务端返回 seq<45 的一页。
+      repoMock.__setMessagesByConversation('c1', makeRange(45, 49));
+
+      serviceMock.__setMessagesHandler(makeServerHandler());
+
+      const sync = createSyncService();
+      await sync.fetchMessages('c1');
+
+      const msgs = useMessageStore.getState().getMessages('c1');
+      // 应连续到 seq=1
+      expect(msgs).toHaveLength(49);
+      const seqs = msgs.map((m) => m.seq);
+      expect(seqs).toEqual(expect.arrayContaining(makeRange(1, 49).map((m) => m.seq)));
+      expect(Math.min(...seqs)).toBe(1);
+      expect(Math.max(...seqs)).toBe(49);
+    });
+
+    it('回填在 MAX_GAP_FILL_ROUNDS 后停止，不卡死', async () => {
+      // 场景：历史巨大（seq 到 10000），每页 MESSAGES_PAGE_SIZE，
+      //       5 轮最多回填 5*20=100 条，无法到顶，应在上限处停止。
+      const TOP = 10000;
+      repoMock.__setMessagesByConversation('c1', makeRange(TOP - 4, TOP, 'top'));
+
+      serviceMock.__setMessagesHandler(makeServerHandler());
+
+      const sync = createSyncService();
+      await sync.fetchMessages('c1');
+
+      const msgs = useMessageStore.getState().getMessages('c1');
+      // 本地 5 + 至多 5*20=100 回填 = 至多 105 条，远小于 10000（证明未无限循环）
+      expect(msgs.length).toBeLessThanOrEqual(5 + MAX_GAP_FILL_ROUNDS * MESSAGES_PAGE_SIZE);
+      expect(msgs.length).toBeGreaterThan(5); // 确实回填了
+      // 仍标记 hydrated（避免反复重试卡首屏）
+      expect(useMessageStore.getState().isHydrated('c1')).toBe(true);
+    });
+
+    it('回填后消息按 seq 严格升序、无重复', async () => {
+      repoMock.__setMessagesByConversation('c1', makeRange(10, 15));
+      serviceMock.__setMessagesHandler(makeServerHandler());
+
+      const sync = createSyncService();
+      await sync.fetchMessages('c1');
+
+      const msgs = useMessageStore.getState().getMessages('c1');
+      const seqs = msgs.map((m) => m.seq);
+      // 严格升序
+      for (let i = 1; i < seqs.length; i++) {
+        expect(seqs[i]).toBeGreaterThan(seqs[i - 1]);
+      }
+      // 无重复 id
+      const ids = msgs.map((m) => m.id);
+      expect(new Set(ids).size).toBe(ids.length);
+    });
+  });
+
+  describe('方案 C：hydrated 门控', () => {
+    it('首次进入走完整管线；重入只做轻量增量同步', async () => {
+      let snapshotCallCount = 0;
+      let beforeSeqCallCount = 0;
+      let localReadCount = 0;
+      // 首次 hydration 的上缺口同步（after_seq=20）返回空；
+      // 重入的增量同步（after_seq=20）返回 seq=21，用于验证重入确实走了增量路径。
+      let afterSeqCallCount = 0;
+
+      const realGetByConversation = repoMock.getByConversation.bind(repoMock);
+      repoMock.getByConversation = async (id: string, limit?: number) => {
+        localReadCount++;
+        return realGetByConversation(id, limit);
+      };
+
+      repoMock.__setMessagesByConversation('c1', makeRange(1, 20));
+
+      serviceMock.__setMessagesHandler(({ afterSeq, beforeSeq }: any) => {
+        if (afterSeq !== undefined) {
+          afterSeqCallCount++;
+          // 仅重入（第 2 次 afterSeq 请求）返回新消息 seq=21
+          if (afterSeqCallCount === 2) {
+            return { messages: [makeMessage('new-21', 21)], total: 0, page: 1, page_size: 20 };
+          }
+          return { messages: [], total: 0, page: 1, page_size: 20 };
+        }
+        if (beforeSeq !== undefined) {
+          beforeSeqCallCount++;
+          // 首次 hydration 本地已连续到 1，理论上不应被调用
+          return { messages: [], total: 0, page: 1, page_size: 20 };
+        }
+        snapshotCallCount++;
+        return { messages: [], total: 0, page: 1, page_size: 20 };
+      });
+
+      const sync = createSyncService();
+
+      // 首次：完整 hydration。本地 1~20 连续到顶，无需回填。
+      const localReadAfterFirst = localReadCount;
+      await sync.fetchMessages('c1');
+      expect(useMessageStore.getState().isHydrated('c1')).toBe(true);
+      expect(useMessageStore.getState().getMessages('c1')).toHaveLength(20);
+      // 本地已连续到顶，回填不应触发服务端 beforeSeq 请求
+      expect(beforeSeqCallCount).toBe(0);
+      expect(snapshotCallCount).toBe(0);
+      expect(localReadCount).toBeGreaterThan(localReadAfterFirst); // 首次读了本地
+
+      // 重入：应只做轻量增量（after_seq=maxSeq=20），不再读本地/回填
+      const localReadBeforeReentry = localReadCount;
+      await sync.fetchMessages('c1');
+      // 重入后增量拿到 seq=21
+      const msgs = useMessageStore.getState().getMessages('c1');
+      expect(msgs).toHaveLength(21);
+      expect(msgs.some((m) => m.seq === 21)).toBe(true);
+      // 重入不应再读本地、不应触发快照或回填
+      expect(localReadCount).toBe(localReadBeforeReentry);
+      expect(snapshotCallCount).toBe(0);
+      expect(beforeSeqCallCount).toBe(0);
+    });
+  });
+
+  describe('并发安全', () => {
+    it('hydration 期间 WS ingest 新消息，最终消息完整不丢失', async () => {
+      // 本地有 seq=1~20，内存初始空
+      repoMock.__setMessagesByConversation('c1', makeRange(1, 20));
+
+      serviceMock.__setMessagesHandler(makeServerHandler());
+
+      const sync = createSyncService();
+      const fetchPromise = sync.fetchMessages('c1');
+
+      // 在 hydration 进行中模拟 WS ingest 一条新消息
+      useMessageStore.getState().addOrReplaceMessage('c1', makeMessage('ws-21', 21));
+
+      await fetchPromise;
+
+      const msgs = useMessageStore.getState().getMessages('c1');
+      const seqs = msgs.map((m) => m.seq);
+      // 本地 1~20 + WS 的 21，共 21 条，无丢失
+      expect(msgs).toHaveLength(21);
+      expect(seqs).toContain(21);
+      expect(new Set(msgs.map((m) => m.id)).size).toBe(msgs.length);
+    });
+  });
+
+  describe('清空聊天记录后重新 hydration', () => {
+    it('clearMessages 清空内存消息 + hydrated 标志，下次进入重新走完整管线', async () => {
+      // 场景：会话已 hydrated 且内存有消息；用户清空记录后重新进入。
+      // 修复前：hydrated 残留 → 走轻量增量（after_seq=旧maxSeq）→ 清空后仍显示旧消息或拉不到新历史。
+      // 修复后：clearMessages 清内存+hydrated → 下次 fetchMessages 重新走完整 hydration。
+      messageRepository.__setMessagesByConversation('c1', makeRange(1, 20));
+      messageService.__setMessagesHandler(makeServerHandler());
+
+      const sync = createSyncService();
+
+      // 首次 hydration
+      await sync.fetchMessages('c1');
+      expect(useMessageStore.getState().isHydrated('c1')).toBe(true);
+      expect(useMessageStore.getState().getMessages('c1')).toHaveLength(20);
+
+      // 用户清空记录（DB + 内存 + hydrated）
+      await messageRepository.clearConversation('c1');
+      useMessageStore.getState().clearMessages('c1');
+      expect(useMessageStore.getState().getMessages('c1')).toHaveLength(0);
+      expect(useMessageStore.getState().isHydrated('c1')).toBe(false);
+
+      // 重新进入：本地 DB 已空（clearConversation 清了 mock 状态），走冷启动快照
+      // 配置服务端快照返回 seq=1~10（模拟清空后服务端重新返回的消息）
+      messageService.__setMessagesHandler(
+        makeServerHandler({ snapshot: makeRange(1, 10) })
+      );
+
+      await sync.fetchMessages('c1');
+
+      const msgs = useMessageStore.getState().getMessages('c1');
+      // 应重新加载到快照消息，而非停留在清空状态
+      expect(msgs).toHaveLength(10);
+      expect(useMessageStore.getState().isHydrated('c1')).toBe(true);
+    });
+  });
+});

src/stores/message/__tests__/mocks/databaseMock.ts

@@ -1,26 +1,125 @@
 /**
  * 测试用 database mock：messageRepository / conversationRepository / userCacheRepository。
- * 仅提供方法存根，单测断言 store 层行为，不验证持久化。
+ *
+ * 默认行为：所有读取返回空、写入为 no-op，与历史行为一致（不破坏现有单测）。
+ * 增强能力：单测可通过 `messageRepository.__setState()` / `messageRepository.__setHandler()`
+ * 注入按 (conversationId, 参数) 维度的响应，用于 hydration 回填管线等需要
+ * 真实数据流的场景。每个 test 的 beforeEach 应调用 `__reset()` 复原默认行为。
  */
 
+// ---- 可配置的内部状态 ----
+
+interface RepositoryMockState {
+  /** 按 `${conversationId}` 存储该会话的全部本地消息（模拟 SQLite 表内容）。 */
+  messagesByConversation: Map<string, any[]>;
+}
+
+const state: RepositoryMockState = {
+  messagesByConversation: new Map(),
+};
+
+function resetState() {
+  state.messagesByConversation.clear();
+}
+
+/**
+ * 让 `getByConversation` / `getBeforeSeq` / `getContiguousRange` 等读取方法
+ * 基于注入的消息集合工作（模拟本地 DB 已落库的数据）。
+ * `messages` 应按 seq 升序提供。
+ */
+function __setMessagesByConversation(conversationId: string, messages: any[]) {
+  state.messagesByConversation.set(conversationId, [...messages].sort((a, b) => a.seq - b.seq));
+}
+
 export const messageRepository = {
+  // ---- 写入：no-op（单测不验证持久化）----
   saveMessage: async () => {},
   saveMessagesBatch: async () => {},
-  getByConversation: async () => [],
+
-  getBeforeSeq: async () => [],
+  // ---- 读取：基于内部状态 ----
+  async getByConversation(conversationId: string, limit = 20) {
+    const all = state.messagesByConversation.get(conversationId) || [];
+    // 模拟 SQL：ORDER BY seq DESC LIMIT，再 reverse() 为升序
+    return [...all].sort((a, b) => b.seq - a.seq).slice(0, limit).reverse();
+  },
+
+  async getBeforeSeq(conversationId: string, beforeSeq: number, limit = 20) {
+    const all = state.messagesByConversation.get(conversationId) || [];
+    return [...all]
+      .filter((m) => m.seq < beforeSeq)
+      .sort((a, b) => b.seq - a.seq)
+      .slice(0, limit)
+      .reverse();
+  },
+
+  async getMaxSeq(conversationId: string) {
+    const all = state.messagesByConversation.get(conversationId) || [];
+    return all.reduce((max, m) => Math.max(max, m.seq || 0), 0);
+  },
+
+  async getMinSeq(conversationId: string) {
+    const all = state.messagesByConversation.get(conversationId) || [];
+    if (all.length === 0) return 0;
+    return all.reduce((min, m) => Math.min(min, m.seq || Infinity), Infinity);
+  },
+
+  async getContiguousRange(conversationId: string) {
+    const all = (state.messagesByConversation.get(conversationId) || [])
+      .filter((m) => m.seq > 0)
+      .map((m) => m.seq)
+      .sort((a, b) => b - a); // DESC
+    if (all.length === 0) return { maxSeq: 0, minSeq: 0 };
+    let maxSeq = all[0];
+    let minSeq = maxSeq;
+    for (let i = 1; i < all.length; i++) {
+      if (all[i] === minSeq - 1) {
+        minSeq = all[i];
+      } else {
+        break;
+      }
+    }
+    return { maxSeq, minSeq };
+  },
+
+  async getCount(conversationId: string) {
+    return (state.messagesByConversation.get(conversationId) || []).length;
+  },
+
+  // ---- 其它：no-op / 空实现 ----
   updateStatus: async () => {},
+  markAsRead: async () => {},
   markConversationAsRead: async () => {},
   delete: async () => {},
-  clearConversation: async () => {},
+  clearConversation: async (conversationId: string) => {
+    state.messagesByConversation.delete(conversationId);
+  },
+  search: async () => [],
+  searchByConversation: async () => [],
+  getStats: async () => ({ totalMessages: 0, totalConversations: 0 }),
+  getLastMessageByConversation: async () => null,
+  getCountBeforeSeq: async () => 0,
+
+  // ---- 测试辅助 ----
+  __reset: resetState,
+  __setState: resetState, // 别名，语义对齐 asyncStorageMock
+  __setMessagesByConversation,
+  __getMessagesByConversation: (conversationId: string) =>
+    state.messagesByConversation.get(conversationId) || [],
 };
 
 export const conversationRepository = {
   getListCache: async () => [],
   saveWithRelated: async () => {},
   delete: async () => {},
+  saveCache: async () => {},
+  getCache: async () => null,
+  updateCacheUnreadCount: async () => {},
+  clearAll: async () => {},
 };
 
 export const userCacheRepository = {
   get: () => null,
   set: () => {},
+  save: async () => {},
+  saveBatch: async () => {},
 };

src/stores/message/__tests__/mocks/messageServiceMock.ts

@@ -1,9 +1,42 @@
 /**
  * 测试用 messageService mock：仅覆盖单测路径调用的方法。
+ *
+ * 默认行为：所有方法返回空/no-op，与历史行为一致（不破坏现有单测）。
+ * 增强能力：`getMessages` 可通过 `__setMessagesHandler` 配置响应分发，
+ * 用于 hydration 回填管线等需要按 (afterSeq/beforeSeq/limit) 返回不同数据的场景。
+ * 每个 test 的 beforeEach 应调用 `__reset()` 复原默认行为。
  */
 
+export type MessageListResponse = { messages: any[]; total: number; page: number; page_size: number };
+
+type GetMessagesHandler = (args: {
+  conversationId: string;
+  afterSeq?: number;
+  beforeSeq?: number;
+  limit?: number;
+}) => MessageListResponse;
+
+const EMPTY_RESPONSE: MessageListResponse = { messages: [], total: 0, page: 1, page_size: 20 };
+
+let getMessagesHandler: GetMessagesHandler | null = null;
+
+function resetHandler() {
+  getMessagesHandler = null;
+}
+
 export const messageService = {
-  getMessages: async () => ({ messages: [], total: 0, page: 1, page_size: 20 }),
+  async getMessages(
+    conversationId: string,
+    afterSeq?: number,
+    beforeSeq?: number,
+    limit?: number
+  ): Promise<MessageListResponse> {
+    if (getMessagesHandler) {
+      return getMessagesHandler({ conversationId, afterSeq, beforeSeq, limit });
+    }
+    return { ...EMPTY_RESPONSE };
+  },
+
   getConversationById: async () => null,
   getUnreadCount: async () => ({ total_unread_count: 0 }),
   getSystemUnreadCount: async () => ({ unread_count: 0 }),
@@ -13,6 +46,12 @@ export const messageService = {
   markAllSystemMessagesRead: async () => {},
   fetchRemoteConversationListPage: async () => ({ items: [], hasMore: false }),
   getSystemMessagesCursor: async () => ({ list: [] }),
+
+  // ---- 测试辅助 ----
+  __reset: resetHandler,
+  __setMessagesHandler: (handler: GetMessagesHandler) => {
+    getMessagesHandler = handler;
+  },
 };
 
-export type MessageListResponse = { messages: any[]; total: number; page: number; page_size: number };
+export type { GetMessagesHandler };

src/stores/message/constants.ts

@@ -20,3 +20,19 @@ export const MAX_FLUSH_ITERATIONS = 3;
  * 防止频繁重连导致的重复同步
  */
 export const MIN_RECONNECT_SYNC_INTERVAL = 1500;
+
+/**
+ * 单次消息拉取/本地读取的页大小。
+ * 对齐现有 loadMoreMessages 的默认 limit，作为 hydration 管线的统一页大小。
+ */
+export const MESSAGES_PAGE_SIZE = 20;
+
+/**
+ * 进入会话时「下缺口回填」的最大循环轮数（对标 Matrix backpagination 上限）。
+ *
+ * 当本地连续区间 [maxSeq, minSeq] 的 minSeq > 1（即最新端往前存在历史缺口）时，
+ * hydration 管线会循环 loadMoreMessages 向前回填，直到连续到顶（minSeq==1）
+ * 或触达本上限。上限是为了防止极端情况（如会话有数万条历史）下首屏卡死，
+ * 剩余缺口由用户上滑 loadMoreHistory 兜底补全。
+ */
+export const MAX_GAP_FILL_ROUNDS = 5;

src/stores/message/services/MessageSyncService.ts

@@ -16,6 +16,10 @@ import {
 import type { IMessageSyncService, MessageManagerConversationListDeps, IUserCacheService } from '../types';
 import { useMessageStore, normalizeConversationId } from '../store';
 import { ReadReceiptManager } from './ReadReceiptManager';
+import {
+  MESSAGES_PAGE_SIZE,
+  MAX_GAP_FILL_ROUNDS,
+} from '../constants';
 
 export class MessageSyncService implements IMessageSyncService {
   private getCurrentUserId: () => string | null;
@@ -216,6 +220,15 @@ export class MessageSyncService implements IMessageSyncService {
 
   /**
    * 获取会话消息（增量同步）— 核心实现
+   *
+   * 【hydration 管线，修复「进入聊天页只显示一条消息、历史被吞」】
+   *
+   * 未 hydrated 的会话走完整管线：
+   *   1. 本地 DB 连续区间补全（方案 A：不信内存零散消息，始终读本地）
+   *   2. 上缺口增量同步（拉 maxSeq 之后的新消息）
+   *   3. 下缺口完整回填（循环 loadMore 直到连续到顶，方案 B）
+   *   4. 标记 hydrated（方案 C）
+   * 已 hydrated 的会话只做轻量增量同步（baseline 此时可信）。
    */
   private async _doFetchMessages(conversationId: string, afterSeq?: number): Promise<void> {
     const store = useMessageStore.getState();
@@ -224,76 +237,16 @@ export class MessageSyncService implements IMessageSyncService {
 
     try {
       if (!afterSeq) {
-        // 先从本地数据库加载（使用 merge 避免 WS 消息丢失）
+        const wasHydrated = store.isHydrated(conversationId);
-        const existingMessagesAtStart = store.getMessages(conversationId);
+        if (wasHydrated) {
-        if (existingMessagesAtStart.length === 0) {
+          // 已完整加载过：baseline 可信，仅做轻量增量同步
-          try {
+          await this.incrementalSyncAbove(conversationId);
-            const localMessages = await messageRepository.getByConversation(conversationId, 20);
-
-            if (localMessages.length > 0) {
-              const formattedMessages: MessageResponse[] = localMessages.map(m => ({
-                id: m.id,
-                conversation_id: m.conversationId,
-                sender_id: m.senderId,
-                seq: m.seq,
-                segments: m.segments || [],
-                status: m.status as any,
-                created_at: m.createdAt,
-              }));
-
-              // 原子性合并本地缓存消息，防止并发 WS 消息丢失
-              store.mergeMessages(conversationId, formattedMessages);
-            }
-          } catch (error) {
-            console.warn('[MessageSyncService] 读取本地消息失败:', error);
-          }
-        }
-
-        // 更新 baseline：合并本地DB后重新计算
-        const currentMessages = store.getMessages(conversationId);
-        const baselineMaxSeq = currentMessages.reduce((max, m) => Math.max(max, m.seq || 0), 0);
-
-        // 策略：有本地数据时仅增量同步，无数据时拉快照
-        if (baselineMaxSeq > 0) {
-          // 增量同步：只拉 baselineMaxSeq 之后的消息
-          try {
-            const incrementalResp = await messageService.getMessages(conversationId, baselineMaxSeq);
-            const newMessages = incrementalResp?.messages || [];
-
-            if (newMessages.length > 0) {
-              store.mergeMessages(conversationId, newMessages);
-
-              messageRepository.saveMessagesBatch(
-                MessageMapper.toCachedMessages(newMessages, conversationId)
-              ).catch(error => {
-                console.error('[MessageSyncService] 保存增量消息到本地失败:', error);
-              });
-            }
-          } catch (error) {
-            console.error('[MessageSyncService] 增量同步失败:', error);
-          }
         } else {
-          // 冷启动：本地无数据，拉快照
+          // 完整 hydration 管线
-          try {
+          await this.runHydrationPipeline(conversationId);
-            const snapshotResp = await messageService.getMessages(conversationId, undefined, undefined, 50);
-            const snapshotMessages = snapshotResp?.messages || [];
-
-            if (snapshotMessages.length > 0) {
-              // 快照来自分页接口，后端返回 DESC 顺序；mergeMessages 内部按 seq ASC 排序并合并
-              store.mergeMessages(conversationId, snapshotMessages);
-
-              messageRepository.saveMessagesBatch(
-                MessageMapper.toCachedMessages(snapshotMessages, conversationId)
-              ).catch(error => {
-                console.error('[MessageSyncService] 保存快照消息到本地失败:', error);
-              });
-            }
-          } catch (error) {
-            console.error('[MessageSyncService] 快照同步失败:', error);
-          }
         }
       } else {
-        // 指定了 afterSeq
+        // 指定了 afterSeq（焦点/重连增量同步）：保持原行为
         const response = await messageService.getMessages(conversationId, afterSeq);
 
         if (response?.messages && response.messages.length > 0) {
@@ -334,6 +287,181 @@ export class MessageSyncService implements IMessageSyncService {
     }
   }
 
+  /**
+   * 轻量增量同步：以当前内存最大 seq 为 baseline，只拉取更新的消息。
+   * 仅对已 hydrated 的会话调用——此时 baseline 是可信的（已通过完整管线加载）。
+   */
+  private async incrementalSyncAbove(conversationId: string): Promise<void> {
+    const store = useMessageStore.getState();
+    const currentMessages = store.getMessages(conversationId);
+    const baselineMaxSeq = currentMessages.reduce((max, m) => Math.max(max, m.seq || 0), 0);
+
+    if (baselineMaxSeq <= 0) {
+      // 已 hydrated 但内存空（如清空后未清 hydrated 的边界）：退回快照兜底
+      await this.fetchSnapshot(conversationId);
+      return;
+    }
+
+    try {
+      const incrementalResp = await messageService.getMessages(conversationId, baselineMaxSeq);
+      const newMessages = incrementalResp?.messages || [];
+      if (newMessages.length > 0) {
+        store.mergeMessages(conversationId, newMessages);
+        messageRepository.saveMessagesBatch(
+          MessageMapper.toCachedMessages(newMessages, conversationId)
+        ).catch(error => {
+          console.error('[MessageSyncService] 增量同步保存失败:', error);
+        });
+      }
+    } catch (error) {
+      console.error('[MessageSyncService] 增量同步失败:', error);
+    }
+  }
+
+  /**
+   * 完整 hydration 管线（方案 A + B + C 的核心）。
+   *
+   * 流程：
+   *   Step 1（A）：无条件读本地 DB 补全内存基线（修复「内存非空就跳过本地」的根因）
+   *   Step 2：上缺口增量同步（拉本地 maxSeq 之后的新消息）
+   *   Step 3（B）：下缺口完整回填循环（向前 loadMore 直到连续到顶或触达上限）
+   *   Step 4（C）：标记 hydrated
+   */
+  private async runHydrationPipeline(conversationId: string): Promise<void> {
+    const store = useMessageStore.getState();
+
+    // ---- Step 1（方案 A）：本地 DB 连续区间补全 ----
+    // 关键修复：移除「内存非空就跳过本地」的错误前提。WS 全局 ingest 会让内存非空，
+    // 但那只是零散最新消息，不代表历史完整。始终用本地 DB 补全基线。
+    try {
+      const localMessages = await messageRepository.getByConversation(conversationId, MESSAGES_PAGE_SIZE);
+      if (localMessages.length > 0) {
+        const formattedMessages: MessageResponse[] = localMessages.map(m => ({
+          id: m.id,
+          conversation_id: m.conversationId,
+          sender_id: m.senderId,
+          seq: m.seq,
+          segments: m.segments || [],
+          status: m.status as any,
+          created_at: m.createdAt,
+        }));
+        // mergeMessages 按 id 去重 + seq 升序，与内存现有 WS 消息安全合并
+        useMessageStore.getState().mergeMessages(conversationId, formattedMessages);
+      }
+    } catch (error) {
+      console.warn('[MessageSyncService] hydration 读取本地消息失败:', error);
+    }
+
+    // ---- Step 2：上缺口增量同步（baselineMaxSeq 之后的新消息）----
+    const messagesAfterLocal = useMessageStore.getState().getMessages(conversationId);
+    const baselineMaxSeq = messagesAfterLocal.reduce((max, m) => Math.max(max, m.seq || 0), 0);
+
+    if (baselineMaxSeq > 0) {
+      try {
+        const incrementalResp = await messageService.getMessages(conversationId, baselineMaxSeq);
+        const newMessages = incrementalResp?.messages || [];
+        if (newMessages.length > 0) {
+          useMessageStore.getState().mergeMessages(conversationId, newMessages);
+          messageRepository.saveMessagesBatch(
+            MessageMapper.toCachedMessages(newMessages, conversationId)
+          ).catch(error => {
+            console.error('[MessageSyncService] hydration 增量同步保存失败:', error);
+          });
+        }
+      } catch (error) {
+        console.error('[MessageSyncService] hydration 增量同步失败:', error);
+      }
+    } else {
+      // 本地与内存均空：冷启动快照兜底
+      await this.fetchSnapshot(conversationId);
+    }
+
+    // ---- Step 3（方案 B）：下缺口完整回填循环 ----
+    // 从最新端往前检查连续性，发现缺口就 loadMore 回填，直到连续到顶（minSeq<=1）
+    // 或触达 MAX_GAP_FILL_ROUNDS 上限（防极端卡死，剩余靠用户上滑兜底）。
+    await this.fillBackwardGaps(conversationId);
+
+    // ---- Step 4（方案 C）：标记 hydrated ----
+    // 无论回填是否完全到顶都标记，避免反复重试卡首屏；用户上滑仍可触发 loadMoreHistory。
+    useMessageStore.getState().markHydrated(conversationId);
+  }
+
+  /**
+   * 冷启动快照：本地与内存均无数据时，拉取最近一页快照。
+   */
+  private async fetchSnapshot(conversationId: string): Promise<void> {
+    try {
+      const snapshotResp = await messageService.getMessages(conversationId, undefined, undefined, 50);
+      const snapshotMessages = snapshotResp?.messages || [];
+      if (snapshotMessages.length > 0) {
+        // 快照来自分页接口，后端返回 DESC 顺序；mergeMessages 内部按 seq ASC 排序并合并
+        useMessageStore.getState().mergeMessages(conversationId, snapshotMessages);
+        messageRepository.saveMessagesBatch(
+          MessageMapper.toCachedMessages(snapshotMessages, conversationId)
+        ).catch(error => {
+          console.error('[MessageSyncService] 保存快照消息到本地失败:', error);
+        });
+      }
+    } catch (error) {
+      console.error('[MessageSyncService] 快照同步失败:', error);
+    }
+  }
+
+  /**
+   * 下缺口回填循环（方案 B）。
+   *
+   * 从当前内存消息的最新端向前找第一个连续缺口，loadMore 一页补上，循环直到：
+   *   - 连续到顶（contiguousMinSeq <= 1），或
+   *   - 触达 MAX_GAP_FILL_ROUNDS 上限，或
+   *   - 某轮回填无新增（已到顶或网络空响应），防死循环。
+   *
+   * 连续区间在「内存消息数组」上计算（回填每轮都 merge 进内存，比每轮查 DB 更高效）。
+   */
+  private async fillBackwardGaps(conversationId: string): Promise<void> {
+    for (let round = 0; round < MAX_GAP_FILL_ROUNDS; round++) {
+      const currentMessages = useMessageStore.getState().getMessages(conversationId);
+      if (currentMessages.length === 0) return;
+
+      const contiguousMinSeq = this.computeContiguousMinSeq(currentMessages);
+      // 连续到顶（minSeq <= 1 表示已到最早消息），无需回填
+      if (contiguousMinSeq <= 1) return;
+
+      const beforeCount = currentMessages.length;
+      try {
+        // 复用现有 loadMoreMessages：本地不足则请求服务端，内部已 merge + 落库
+        await this.loadMoreMessages(conversationId, contiguousMinSeq, MESSAGES_PAGE_SIZE);
+      } catch (error) {
+        console.error('[MessageSyncService] 回填历史缺口失败:', error);
+        return; // 本轮失败，停止回填（已加载部分正常展示）
+      }
+
+      // 本轮无新增消息：已到顶或服务端无更多，停止回填
+      const afterCount = useMessageStore.getState().getMessages(conversationId).length;
+      if (afterCount <= beforeCount) return;
+    }
+  }
+
+  /**
+   * 计算内存消息数组从最大 seq 往下的连续区间下沿（contiguousMinSeq）。
+   * 遇到第一个缺口即停。返回值 <= 1 表示已连续到顶。
+   */
+  private computeContiguousMinSeq(messages: MessageResponse[]): number {
+    if (messages.length === 0) return 0;
+    // 按 seq 降序遍历找连续段
+    const sorted = [...messages].sort((a, b) => b.seq - a.seq);
+    let expected = sorted[0].seq;
+    for (let i = 1; i < sorted.length; i++) {
+      if (sorted[i].seq === expected - 1) {
+        expected = sorted[i].seq;
+      } else if (sorted[i].seq < expected - 1) {
+        // 遇到缺口
+        break;
+      }
+      // sorted[i].seq === expected（重复，幂等已去重，理论不出现）→ 跳过
+    }
+    return expected;
+  }
+
   /**
    * 加载更多历史消息
    */

src/stores/message/store.ts

@@ -42,6 +42,17 @@ export interface MessageState {
   isLoadingConversations: boolean;
   loadingMessagesSet: Set<string>;
 
+  /**
+   * 已完成完整 hydration（连续区间补全）的会话集合。
+   *
+   * 区别于 loadingMessagesSet（仅表示「正在拉取」），hydrated 标志语义是
+   * 「该会话的消息历史已通过完整管线加载到本地连续，后续进入只需轻量增量同步」。
+   * 用于修复「从消息栏进入聊天页只显示零散 WS 消息、历史被吞」的问题：
+   * WS 全局 ingest 会让内存非空，但那不代表历史已连续加载，必须以 hydrated
+   * 作为「是否需要执行完整 hydration 管线」的判据。
+   */
+  hydratedConversations: Set<string>;
+
   // 初始化状态
   isInitialized: boolean;
 
@@ -96,6 +107,14 @@ export interface MessageActions {
    */
   addOrReplaceMessage: (conversationId: string, message: MessageResponse) => boolean;
   removeMessage: (conversationId: string, messageId: string) => void;
+  /**
+   * 清空某会话的内存消息并清除 hydrated 标志。
+   *
+   * 用于「清空聊天记录」场景：调用方先 messageRepository.clearConversation 清 DB，
+   * 再调本 action 同步内存状态，保证内存/DB/hydrated 三者一致。
+   * 清除 hydrated 后，下次进入会话会重新走完整 hydration 管线（重新从服务端加载）。
+   */
+  clearMessages: (conversationId: string) => void;
   patchMessages: (conversationId: string, patches: Map<string, Partial<MessageResponse>>) => void;
   setUnreadCount: (total: number, system: number) => void;
   /**
@@ -116,6 +135,15 @@ export interface MessageActions {
   setCurrentConversation: (conversationId: string | null) => void;
   setLoading: (loading: boolean) => void;
   setLoadingMessages: (conversationId: string, loading: boolean) => void;
+  /**
+   * 是否已完成完整 hydration（连续区间补全）。
+   * hydration 管线据此门控：未 hydrated 才执行完整的本地补全 + 缺口回填。
+   */
+  isHydrated: (conversationId: string) => boolean;
+  /** 标记某会话已完成完整 hydration（幂等，已存在则不变）。 */
+  markHydrated: (conversationId: string) => void;
+  /** 清除某会话的 hydrated 标志（删除会话/清空记录时调用，避免残留）。 */
+  clearHydrated: (conversationId: string) => void;
   setTypingUsers: (groupId: string, users: string[]) => void;
   setMutedStatus: (groupId: string, isMuted: boolean) => void;
   setNotificationMuted: (conversationId: string, notificationMuted: boolean) => void;
@@ -157,6 +185,7 @@ const initialState: MessageState = {
   currentConversationId: null,
   isLoadingConversations: false,
   loadingMessagesSet: new Set(),
+  hydratedConversations: new Set(),
   isInitialized: false,
   typingUsersMap: new Map(),
   mutedStatusMap: new Map(),
@@ -226,6 +255,7 @@ export const useMessageStore = create<MessageStore>((set, get) => ({
       currentConversationId: state.currentConversationId,
       isLoadingConversations: state.isLoadingConversations,
       loadingMessagesSet: state.loadingMessagesSet,
+      hydratedConversations: state.hydratedConversations,
       isInitialized: state.isInitialized,
       typingUsersMap: state.typingUsersMap,
       mutedStatusMap: state.mutedStatusMap,
@@ -339,6 +369,10 @@ export const useMessageStore = create<MessageStore>((set, get) => ({
       const newLoadingMessagesSet = new Set(state.loadingMessagesSet);
       newLoadingMessagesSet.delete(normalizedId);
 
+      // 同步清除 hydrated 标志，避免会话重建后误判为「已完整加载」
+      const newHydratedConversations = new Set(state.hydratedConversations);
+      newHydratedConversations.delete(normalizedId);
+
       const newTotalUnreadCount = Math.max(0, state.totalUnreadCount - removedUnread);
 
       const newCurrentConversationId = state.currentConversationId === normalizedId
@@ -354,6 +388,7 @@ export const useMessageStore = create<MessageStore>((set, get) => ({
         totalUnreadCount: newTotalUnreadCount,
         currentConversationId: newCurrentConversationId,
         loadingMessagesSet: newLoadingMessagesSet,
+        hydratedConversations: newHydratedConversations,
       };
     });
   },
@@ -421,6 +456,25 @@ export const useMessageStore = create<MessageStore>((set, get) => ({
     });
   },
 
+  clearMessages: (conversationId: string) => {
+    const normalizedId = normalizeConversationId(conversationId);
+    set(state => {
+      const newMessagesMap = new Map(state.messagesMap);
+      newMessagesMap.delete(normalizedId);
+      // 同步清除 hydrated 标志，下次进入会重新走完整 hydration 管线
+      const newHydratedConversations = new Set(state.hydratedConversations);
+      newHydratedConversations.delete(normalizedId);
+      // 同步清除 loading 锁，避免残留锁阻断后续 fetchMessages
+      const newLoadingMessagesSet = new Set(state.loadingMessagesSet);
+      newLoadingMessagesSet.delete(normalizedId);
+      return {
+        messagesMap: newMessagesMap,
+        hydratedConversations: newHydratedConversations,
+        loadingMessagesSet: newLoadingMessagesSet,
+      };
+    });
+  },
+
   /**
    * 原子性 patch：按消息 ID 更新指定字段（如 sender、status）
    * 不影响其他消息，不丢失并发写入
@@ -513,6 +567,30 @@ export const useMessageStore = create<MessageStore>((set, get) => ({
     });
   },
 
+  isHydrated: (conversationId: string) => {
+    return get().hydratedConversations.has(normalizeConversationId(conversationId));
+  },
+
+  markHydrated: (conversationId: string) => {
+    const normalizedId = normalizeConversationId(conversationId);
+    set(state => {
+      if (state.hydratedConversations.has(normalizedId)) return state;
+      const next = new Set(state.hydratedConversations);
+      next.add(normalizedId);
+      return { hydratedConversations: next };
+    });
+  },
+
+  clearHydrated: (conversationId: string) => {
+    const normalizedId = normalizeConversationId(conversationId);
+    set(state => {
+      if (!state.hydratedConversations.has(normalizedId)) return state;
+      const next = new Set(state.hydratedConversations);
+      next.delete(normalizedId);
+      return { hydratedConversations: next };
+    });
+  },
+
   setTypingUsers: (groupId: string, users: string[]) => {
     set(state => {
       const newTypingUsersMap = new Map(state.typingUsersMap);
@@ -630,6 +708,7 @@ export const useMessageStore = create<MessageStore>((set, get) => ({
       conversations: new Map(),
       messagesMap: new Map(),
       loadingMessagesSet: new Set(),
+      hydratedConversations: new Set(),
       typingUsersMap: new Map(),
       mutedStatusMap: new Map(),
       notificationMutedMap: new Map(),