fix(merge): preserve messages during deduplication

Carry platform and reply IDs through merge and shared import paths so stable IDs drive deduplication.

Keep unrelated chats isolated, bridge only verified overlapping sources, and backfill stable IDs onto retained fallback copies.

packages/http-routes/src/routes/web/merge.test.ts

@@ -81,3 +81,98 @@ test('forwards the session gap threshold when importing merged output', async (t
   assert.equal(response.json().sessionId, 'merged-session')
   assert.deepEqual(receivedOptions, { sessionGapThreshold: 7200 })
 })
+
+test('keeps every distinct platform message when merging a source with its superset', async (t) => {
+  const rootDir = makeTempDir()
+  t.after(() => fs.rmSync(rootDir, { recursive: true, force: true }))
+
+  const pathProvider = createPathProvider(rootDir)
+  const dbManager = new DatabaseManager(pathProvider, { nativeBinding, allowMissingRuntimeForTests: true })
+  const mergeSessionCache = new MergeSessionCache(pathProvider, { nativeBinding })
+
+  const makeMessages = (count: number) =>
+    Array.from({ length: count }, (_, index) => {
+      const fallbackCollisionIndex = index < 10 ? Math.floor(index / 2) : index
+      return {
+        platformMessageId: `message-${index}`,
+        senderPlatformId: 'qq-user-1',
+        senderAccountName: 'Alice',
+        timestamp: 1_700_000_000 + fallbackCollisionIndex,
+        type: 0 as const,
+        content: index < 10 ? `same-second-content-${fallbackCollisionIndex}` : `message-content-${index}`,
+        replyToMessageId: index === 11 ? 'message-10' : undefined,
+      }
+    })
+
+  const createSession = (sessionId: string, messageCount: number): void => {
+    const db = dbManager.openRawSessionDatabase(sessionId, { create: true, initializeChatTables: true })
+    db.prepare('INSERT INTO meta (name, platform, type, imported_at) VALUES (?, ?, ?, ?)').run(
+      'Source',
+      'qq',
+      'private',
+      1_700_000_000
+    )
+    const memberResult = db
+      .prepare('INSERT INTO member (platform_id, account_name) VALUES (?, ?)')
+      .run('qq-user-1', 'Alice')
+    const insertMessage = db.prepare(
+      `INSERT INTO message (
+         sender_id, sender_account_name, ts, type, content, platform_message_id, reply_to_message_id
+       ) VALUES (?, ?, ?, ?, ?, ?, ?)`
+    )
+    for (const message of makeMessages(messageCount)) {
+      insertMessage.run(
+        memberResult.lastInsertRowid,
+        message.senderAccountName,
+        message.timestamp,
+        message.type,
+        message.content,
+        message.platformMessageId,
+        message.replyToMessageId || null
+      )
+    }
+    db.close()
+  }
+
+  createSession('older', 100)
+  createSession('newer', 120)
+
+  const app = Fastify()
+  registerMergeRoutes(app, { dbManager, mergeSessionCache })
+  await app.ready()
+  t.after(() => app.close())
+
+  const exportResponse = await app.inject({
+    method: 'POST',
+    url: '/_web/sessions/export-for-merge',
+    payload: { sessionIds: ['older', 'newer'] },
+  })
+  assert.equal(exportResponse.statusCode, 200)
+  const handles = exportResponse.json().handles as Array<{ handle: string }>
+
+  const conflictResponse = await app.inject({
+    method: 'POST',
+    url: '/_web/merge/conflicts',
+    payload: { handles: handles.map(({ handle }) => handle) },
+  })
+  assert.equal(conflictResponse.statusCode, 200)
+  assert.equal(conflictResponse.json().totalMessages, 120)
+
+  const response = await app.inject({
+    method: 'POST',
+    url: '/_web/merge/execute',
+    payload: {
+      handles: handles.map(({ handle }) => handle),
+      outputName: 'Merged',
+    },
+  })
+
+  assert.equal(response.statusCode, 200)
+  const messages = response.json().data.messages as Array<{
+    platformMessageId?: string
+    replyToMessageId?: string
+  }>
+  assert.equal(messages.length, 120)
+  assert.equal(new Set(messages.map((message) => message.platformMessageId)).size, 120)
+  assert.equal(messages.find((message) => message.platformMessageId === 'message-11')?.replyToMessageId, 'message-10')
+})

packages/http-routes/src/routes/web/merge.ts

@@ -251,12 +251,14 @@ export function registerMergeRoutes(server: FastifyInstance, ctx: MergeRoutesCon
       )
       writer.writeMessages(
         exported.messages.map((msg) => ({
+          platformMessageId: msg.platformMessageId,
           senderPlatformId: msg.sender,
           senderAccountName: msg.accountName,
           senderGroupNickname: msg.groupNickname,
           timestamp: msg.timestamp,
           type: msg.type,
           content: msg.content,
+          replyToMessageId: msg.replyToMessageId,
         }))
       )
       writer.finish()

packages/node-runtime/src/import/streaming-importer.ts

@@ -1025,8 +1025,10 @@ export async function streamParseFileInfo(
     'INSERT OR IGNORE INTO member (platform_id, account_name, group_nickname, avatar) VALUES (?, ?, ?, ?)'
   )
   const insertMessage = db.prepare(
-    `INSERT INTO message (sender_platform_id, sender_account_name, sender_group_nickname, timestamp, type, content)
-     VALUES (?, ?, ?, ?, ?, ?)`
+    `INSERT INTO message (
+       platform_message_id, sender_platform_id, sender_account_name, sender_group_nickname,
+       timestamp, type, content, reply_to_message_id
+     ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
   )
 
   let meta: ParsedMeta = { name: 'Unknown', platform: formatFeature.platform, type: 0 as any }
@@ -1076,12 +1078,14 @@ export async function streamParseFileInfo(
           }
 
           insertMessage.run(
+            msg.platformMessageId || null,
             msg.senderPlatformId,
             msg.senderAccountName || null,
             msg.senderGroupNickname || null,
             msg.timestamp,
             msg.type,
-            msg.content || null
+            msg.content || null,
+            msg.replyToMessageId || null
           )
           messageCount++
         }

packages/node-runtime/src/merger/__tests__/merger.test.ts

@@ -1,17 +1,22 @@
 import { describe, it } from 'node:test'
 import assert from 'node:assert/strict'
 import Database from 'better-sqlite3'
-import { checkConflictsFromSources, buildMergedOutput, serializeChatLabToJsonl } from '../index'
+import {
+  checkConflictsFromSources,
+  buildMergedOutput,
+  serializeChatLabToJsonl,
+  type MergerInputMessage,
+} from '../index'
 import { exportSessionToJson } from '../temp-db'
 import type { MergerDataSource, MergerSourceMeta } from '../index'
-import type { MergerMember, MergerMessage } from '@openchatlab/core'
+import type { MergerMember } from '@openchatlab/core'
 import { CHATLAB_FORMAT_VERSION } from '@openchatlab/shared-types'
 import { BetterSqliteAdapter } from '../../better-sqlite3-adapter'
 
 function createMockSource(
   meta: MergerSourceMeta,
   members: MergerMember[],
-  messages: MergerMessage[]
+  messages: MergerInputMessage[]
 ): MergerDataSource {
   return {
     getMeta: () => meta,
@@ -127,6 +132,214 @@ describe('buildMergedOutput', () => {
     )
     assert.equal(result.chatLabData.meta.platform, 'mixed')
   })
+
+  it('keeps same-platform message IDs separate across unrelated private chats', () => {
+    const first = createMockSource(
+      { name: 'Alice', platform: 'telegram', type: 'private' },
+      [{ platformId: 'owner' }, { platformId: 'alice' }],
+      [
+        {
+          platformMessageId: '1',
+          senderPlatformId: 'alice',
+          timestamp: 100,
+          type: 0,
+          content: 'first chat',
+        },
+      ]
+    )
+    const second = createMockSource(
+      { name: 'Bob', platform: 'telegram', type: 'private' },
+      [{ platformId: 'owner' }, { platformId: 'bob' }],
+      [
+        {
+          platformMessageId: '1',
+          senderPlatformId: 'bob',
+          timestamp: 200,
+          type: 0,
+          content: 'second chat',
+        },
+      ]
+    )
+
+    const result = buildMergedOutput(
+      [
+        { source: first, filename: 'alice.json' },
+        { source: second, filename: 'bob.json' },
+      ],
+      'Merged'
+    )
+
+    const conflicts = checkConflictsFromSources([
+      { source: first, filename: 'alice.json' },
+      { source: second, filename: 'bob.json' },
+    ])
+    assert.equal(conflicts.totalMessages, 2)
+    assert.equal(result.chatLabData.messages.length, 2)
+    assert.equal(new Set(result.chatLabData.messages.map((message) => message.platformMessageId)).size, 2)
+  })
+
+  it('deduplicates overlapping private exports when their observed members differ', () => {
+    const meta: MergerSourceMeta = { name: 'Alice', platform: 'qq', type: 'private' }
+    const sharedMessage: MergerInputMessage = {
+      platformMessageId: 'message-1',
+      senderPlatformId: 'alice',
+      timestamp: 100,
+      type: 0,
+      content: 'shared',
+    }
+    const incomingOnly = createMockSource(meta, [{ platformId: 'alice' }], [sharedMessage])
+    const bothSides = createMockSource(
+      meta,
+      [{ platformId: 'owner' }, { platformId: 'alice' }],
+      [
+        sharedMessage,
+        {
+          platformMessageId: 'message-2',
+          senderPlatformId: 'owner',
+          timestamp: 200,
+          type: 0,
+          content: 'reply',
+        },
+      ]
+    )
+    const sources = [
+      { source: incomingOnly, filename: 'old.json' },
+      { source: bothSides, filename: 'new.json' },
+    ]
+
+    const conflicts = checkConflictsFromSources(sources)
+    const result = buildMergedOutput(sources, 'Merged')
+
+    assert.equal(conflicts.totalMessages, 2)
+    assert.equal(result.chatLabData.messages.length, 2)
+    assert.deepEqual(
+      result.chatLabData.messages.map((message) => message.platformMessageId),
+      ['message-1', 'message-2']
+    )
+  })
+
+  it('deduplicates the same group when a newer export has a different name and more members', () => {
+    const oldMeta: MergerSourceMeta = { name: 'Old project', platform: 'qq', type: 'group' }
+    const newMeta: MergerSourceMeta = { name: 'Renamed project', platform: 'qq', type: 'group' }
+    const sharedMessage: MergerInputMessage = {
+      platformMessageId: 'message-1',
+      senderPlatformId: 'alice',
+      timestamp: 100,
+      type: 0,
+      content: 'shared',
+    }
+    const oldExport = createMockSource(oldMeta, [{ platformId: 'alice' }, { platformId: 'bob' }], [sharedMessage])
+    const newExport = createMockSource(
+      newMeta,
+      [{ platformId: 'alice' }, { platformId: 'bob' }, { platformId: 'carol' }],
+      [
+        sharedMessage,
+        {
+          platformMessageId: 'message-2',
+          replyToMessageId: 'message-1',
+          senderPlatformId: 'carol',
+          timestamp: 200,
+          type: 0,
+          content: 'new',
+        },
+      ]
+    )
+    const sources = [
+      { source: oldExport, filename: 'old.json' },
+      { source: newExport, filename: 'new.json' },
+    ]
+
+    const conflicts = checkConflictsFromSources(sources)
+    const result = buildMergedOutput(sources, 'Merged')
+
+    assert.equal(conflicts.totalMessages, 2)
+    assert.equal(result.chatLabData.messages.length, 2)
+    assert.deepEqual(
+      result.chatLabData.messages.map((message) => message.platformMessageId),
+      ['message-1', 'message-2']
+    )
+    assert.equal(result.chatLabData.messages[1].replyToMessageId, result.chatLabData.messages[0].platformMessageId)
+  })
+
+  it('keeps the same message ID separate across unrelated groups with the same name', () => {
+    const meta: MergerSourceMeta = { name: 'Project', platform: 'qq', type: 'group' }
+    const first = createMockSource(
+      meta,
+      [{ platformId: 'alice' }],
+      [
+        {
+          platformMessageId: 'message-1',
+          senderPlatformId: 'alice',
+          timestamp: 100,
+          type: 0,
+          content: 'first group',
+        },
+      ]
+    )
+    const second = createMockSource(
+      meta,
+      [{ platformId: 'bob' }],
+      [
+        {
+          platformMessageId: 'message-1',
+          senderPlatformId: 'bob',
+          timestamp: 200,
+          type: 0,
+          content: 'second group',
+        },
+      ]
+    )
+    const sources = [
+      { source: first, filename: 'first.json' },
+      { source: second, filename: 'second.json' },
+    ]
+
+    const conflicts = checkConflictsFromSources(sources)
+    const result = buildMergedOutput(sources, 'Merged')
+
+    assert.equal(conflicts.totalMessages, 2)
+    assert.equal(result.chatLabData.messages.length, 2)
+    assert.equal(new Set(result.chatLabData.messages.map((message) => message.platformMessageId)).size, 2)
+  })
+
+  it('backfills a stable ID when an ID-bearing copy bridges a fallback-only message', () => {
+    const meta: MergerSourceMeta = { name: 'Chat', platform: 'qq', type: 'private' }
+    const members = [{ platformId: 'owner' }, { platformId: 'alice' }, { platformId: 'bob' }]
+    const fallbackSource = createMockSource(meta, members, [
+      { senderPlatformId: 'alice', timestamp: 100, type: 0, content: 'root' },
+    ])
+    const idSource = createMockSource(meta, members, [
+      {
+        platformMessageId: 'message-1',
+        senderPlatformId: 'alice',
+        timestamp: 100,
+        type: 0,
+        content: 'root',
+      },
+      {
+        platformMessageId: 'message-2',
+        replyToMessageId: 'message-1',
+        senderPlatformId: 'bob',
+        timestamp: 101,
+        type: 0,
+        content: 'reply',
+      },
+    ])
+
+    const result = buildMergedOutput(
+      [
+        { source: fallbackSource, filename: 'old.json' },
+        { source: idSource, filename: 'new.json' },
+      ],
+      'Merged'
+    )
+
+    assert.equal(result.chatLabData.messages.length, 2)
+    const root = result.chatLabData.messages.find((message) => message.content === 'root')
+    const reply = result.chatLabData.messages.find((message) => message.content === 'reply')
+    assert.equal(root?.platformMessageId, 'message-1')
+    assert.equal(reply?.replyToMessageId, root?.platformMessageId)
+  })
 })
 
 describe('serializeChatLabToJsonl', () => {
@@ -182,22 +395,28 @@ describe('exportSessionToJson', () => {
         avatar TEXT
       );
       CREATE TABLE message (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
         sender_id INTEGER NOT NULL,
         sender_account_name TEXT,
         sender_group_nickname TEXT,
         ts INTEGER NOT NULL,
         type INTEGER NOT NULL,
-        content TEXT
+        content TEXT,
+        platform_message_id TEXT,
+        reply_to_message_id TEXT
       );
       INSERT INTO meta (name, platform, type) VALUES ('Test', 'qq', 'group');
       INSERT INTO member (id, platform_id, account_name) VALUES (1, 'alice', 'Alice');
-      INSERT INTO message (sender_id, sender_account_name, ts, type, content)
-      VALUES (1, 'Alice', 100, 0, 'Hello');
+      INSERT INTO message (
+        sender_id, sender_account_name, ts, type, content, platform_message_id, reply_to_message_id
+      ) VALUES (1, 'Alice', 100, 0, 'Hello', 'message-1', 'message-0');
     `)
 
     try {
       const exported = exportSessionToJson(new BetterSqliteAdapter(rawDb))
       assert.equal(exported.chatlab.version, CHATLAB_FORMAT_VERSION)
+      assert.equal(exported.messages[0].platformMessageId, 'message-1')
+      assert.equal(exported.messages[0].replyToMessageId, 'message-0')
     } finally {
       rawDb.close()
     }

packages/node-runtime/src/merger/index.ts

@@ -7,10 +7,10 @@
 
 import {
   getCollidingPlatformIds,
+  getCollidingPlatformIdsFromMessages,
   normalizePlatformId,
   detectConflictsInMessages,
   mergeMergerMembers,
-  generateMessageKey,
   type MergerMember,
   type MergerMessage,
   type ConflictCheckResult,
@@ -18,6 +18,21 @@ import {
   type MergedMessage,
 } from '@openchatlab/core'
 import { CHATLAB_FORMAT_VERSION } from '@openchatlab/shared-types'
+import {
+  createMessageDedupState,
+  generateFallbackMessageKey,
+  registerMessageAndCheckDuplicate,
+} from '../import/message-deduplicator'
+
+export interface MergerInputMessage extends MergerMessage {
+  platformMessageId?: string
+  replyToMessageId?: string
+}
+
+export interface ChatLabMergedMessage extends MergedMessage {
+  platformMessageId?: string
+  replyToMessageId?: string
+}
 
 // ==================== Data source abstraction ====================
 
@@ -29,6 +44,113 @@ export interface MergerSourceMeta {
   groupAvatar?: string
 }
 
+function createMessageScope(meta: MergerSourceMeta | null, members: MergerMember[], sourceIndex: number): string {
+  const platform = meta?.platform || 'unknown'
+  const type = meta?.type || 'unknown'
+
+  if (meta?.groupId) return JSON.stringify([platform, type, 'group', meta.groupId])
+
+  // A group name and its membership can both collide or change over time. Keep
+  // the source isolated until exact overlapping messages prove it is the same chat.
+  if (type === 'group') return JSON.stringify([platform, type, 'source', sourceIndex])
+
+  const memberIds = [...new Set(members.map((member) => member.platformId))].sort()
+  if (memberIds.length > 0) {
+    if (type === 'private') return JSON.stringify([platform, type, 'members', memberIds])
+    return JSON.stringify([platform, type, 'name-members', meta?.name || '', memberIds])
+  }
+
+  // Without a stable group ID or participant set, sharing message IDs across
+  // sources could silently merge unrelated chats. Keep those source-local.
+  return JSON.stringify([platform, type, 'source', sourceIndex])
+}
+
+function createMessageBridgeFamily(meta: MergerSourceMeta | null): string | undefined {
+  if (!meta || meta.groupId || (meta.type !== 'group' && meta.type !== 'private')) return undefined
+  // The exact message signature proves overlap; the editable chat name must
+  // not prevent two exports of the same chat from sharing a scope.
+  return JSON.stringify([meta.platform || 'unknown', meta.type])
+}
+
+function createMessageIdNormalizer(
+  scopes: Iterable<string>
+): (id: string | undefined, scope: string) => string | undefined {
+  const uniqueScopes = [...new Set(scopes)]
+  const scopeIndexes = new Map(uniqueScopes.map((scope, index) => [scope, index]))
+  const namespaceMessageIds = uniqueScopes.length > 1
+
+  return (id, scope) => {
+    if (!id || !namespaceMessageIds) return id
+    return `__chatlab_message_scope__${scopeIndexes.get(scope) ?? 0}__${encodeURIComponent(id)}`
+  }
+}
+
+interface MessageScopeContext {
+  sourceIndex: number
+  messageScope: string
+  messageBridgeFamily?: string
+}
+
+function createMessageScopeResolver(contexts: Iterable<MessageScopeContext>): {
+  addMessage: (message: MergerInputMessage, context: MessageScopeContext) => void
+  complete: () => void
+  resolve: (sourceIndex: number) => string
+} {
+  const contextBySource = new Map<number, MessageScopeContext>()
+  const parents = new Map<number, number>()
+  const firstSourceBySignature = new Map<string, number>()
+
+  for (const context of contexts) {
+    contextBySource.set(context.sourceIndex, context)
+    parents.set(context.sourceIndex, context.sourceIndex)
+  }
+
+  const find = (sourceIndex: number): number => {
+    const parent = parents.get(sourceIndex) ?? sourceIndex
+    if (parent === sourceIndex) return sourceIndex
+    const root = find(parent)
+    parents.set(sourceIndex, root)
+    return root
+  }
+
+  const union = (firstSourceIndex: number, secondSourceIndex: number): void => {
+    const firstRoot = find(firstSourceIndex)
+    const secondRoot = find(secondSourceIndex)
+    if (firstRoot === secondRoot) return
+
+    const root = Math.min(firstRoot, secondRoot)
+    parents.set(firstRoot === root ? secondRoot : firstRoot, root)
+  }
+
+  return {
+    addMessage(message, context) {
+      if (!context.messageBridgeFamily || !message.platformMessageId) return
+
+      const fallbackKey = generateFallbackMessageKey({
+        timestamp: message.timestamp,
+        senderPlatformId: message.senderPlatformId,
+        type: message.type,
+        content: message.content ?? null,
+        replyToMessageId: message.replyToMessageId,
+      })
+      const signature = JSON.stringify([context.messageBridgeFamily, message.platformMessageId, fallbackKey])
+      const firstSourceIndex = firstSourceBySignature.get(signature)
+      if (firstSourceIndex === undefined) {
+        firstSourceBySignature.set(signature, context.sourceIndex)
+        return
+      }
+      union(firstSourceIndex, context.sourceIndex)
+    },
+    complete() {
+      firstSourceBySignature.clear()
+    },
+    resolve(sourceIndex) {
+      const rootContext = contextBySource.get(find(sourceIndex))
+      return rootContext?.messageScope ?? JSON.stringify(['unknown', 'source', sourceIndex])
+    },
+  }
+}
+
 /**
  * Abstract data source for merger input.
  * In Electron, this wraps TempDbReader. Other platforms can implement
@@ -38,7 +160,7 @@ export interface MergerDataSource {
   getMeta(): MergerSourceMeta | null
   getMembers(): MergerMember[]
   getMessageCount(): number
-  streamMessages(batchSize: number, callback: (messages: MergerMessage[]) => void): void
+  streamMessages(batchSize: number, callback: (messages: MergerInputMessage[]) => void): void
 }
 
 // ==================== Output types ====================
@@ -69,7 +191,7 @@ export interface ChatLabOutput {
   chatlab: ChatLabHeader
   meta: ChatLabMeta
   members: MergedMember[]
-  messages: MergedMessage[]
+  messages: ChatLabMergedMessage[]
 }
 
 // ==================== Conflict checking ====================
@@ -77,19 +199,64 @@ export interface ChatLabOutput {
 export function checkConflictsFromSources(
   dataSources: Array<{ source: MergerDataSource; filename: string }>
 ): ConflictCheckResult {
-  const allMessages: Array<{ msg: MergerMessage; source: string; platform: string }> = []
-
-  for (const { source, filename } of dataSources) {
+  const preparedSources = dataSources.map(({ source, filename }, sourceIndex) => {
     const meta = source.getMeta()
-    const platform = meta?.platform || 'unknown'
+    const members = source.getMembers()
+    return {
+      source,
+      filename,
+      sourceIndex,
+      platform: meta?.platform || 'unknown',
+      messageScope: createMessageScope(meta, members, sourceIndex),
+      messageBridgeFamily: createMessageBridgeFamily(meta),
+    }
+  })
+  const allMessages: Array<{
+    msg: MergerInputMessage
+    source: string
+    sourceIndex: number
+    platform: string
+    messageScope: string
+    messageBridgeFamily?: string
+  }> = []
+
+  for (const { source, filename, sourceIndex, platform, messageScope, messageBridgeFamily } of preparedSources) {
     source.streamMessages(10000, (messages) => {
       for (const msg of messages) {
-        allMessages.push({ msg, source: filename, platform })
+        allMessages.push({ msg, source: filename, sourceIndex, platform, messageScope, messageBridgeFamily })
       }
     })
   }
 
-  return detectConflictsInMessages(allMessages)
+  const result = detectConflictsInMessages(allMessages)
+  const collidingIds = getCollidingPlatformIdsFromMessages(allMessages)
+  const scopeResolver = createMessageScopeResolver(preparedSources)
+  for (const message of allMessages) scopeResolver.addMessage(message.msg, message)
+  scopeResolver.complete()
+  const normalizeMessageId = createMessageIdNormalizer(
+    preparedSources.map(({ sourceIndex }) => scopeResolver.resolve(sourceIndex))
+  )
+  const dedupState = createMessageDedupState()
+  let totalMessages = 0
+
+  for (const { msg, sourceIndex, platform } of allMessages) {
+    const messageScope = scopeResolver.resolve(sourceIndex)
+    const senderPlatformId = normalizePlatformId(msg.senderPlatformId, platform, collidingIds)
+    const duplicate = registerMessageAndCheckDuplicate(
+      {
+        platformMessageId: normalizeMessageId(msg.platformMessageId, messageScope),
+        timestamp: msg.timestamp,
+        senderPlatformId,
+        type: msg.type,
+        content: msg.content ?? null,
+        replyToMessageId: normalizeMessageId(msg.replyToMessageId, messageScope),
+      },
+      dedupState
+    )
+    if (!duplicate) totalMessages++
+  }
+
+  return { ...result, totalMessages }
 }
 
 // ==================== Merge orchestration ====================
@@ -108,12 +275,19 @@ export function buildMergedOutput(
   dataSources: Array<{ source: MergerDataSource; filename: string }>,
   outputName: string
 ): MergeOrchestrationResult {
-  const metas = dataSources.map(({ source, filename }) => ({
-    meta: source.getMeta(),
-    members: source.getMembers(),
-    filename,
-    source,
-  }))
+  const metas = dataSources.map(({ source, filename }, sourceIndex) => {
+    const meta = source.getMeta()
+    const members = source.getMembers()
+    return {
+      meta,
+      members,
+      filename,
+      source,
+      sourceIndex,
+      messageScope: createMessageScope(meta, members, sourceIndex),
+      messageBridgeFamily: createMessageBridgeFamily(meta),
+    }
+  })
 
   const collidingIds = getCollidingPlatformIds(
     metas.map(({ meta, members }) => ({
@@ -130,26 +304,71 @@ export function buildMergedOutput(
     collidingIds
   )
 
-  // Streaming dedup: process batches to avoid loading all into memory
-  const seenKeys = new Set<string>()
-  const mergedMessages: MergedMessage[] = []
+  const uniquePlatforms = [...new Set(metas.map(({ meta }) => meta?.platform || 'unknown'))]
+  const scopeResolver = createMessageScopeResolver(metas)
+  for (const context of metas) {
+    if (!context.messageBridgeFamily) continue
+    context.source.streamMessages(10000, (messages) => {
+      for (const message of messages) scopeResolver.addMessage(message, context)
+    })
+  }
+  scopeResolver.complete()
+  const normalizeMessageId = createMessageIdNormalizer(
+    metas.map(({ sourceIndex }) => scopeResolver.resolve(sourceIndex))
+  )
 
-  for (const { source, meta } of metas) {
+  // Streaming dedup: prefer stable platform message IDs and use the content
+  // fingerprint only for messages that do not have one.
+  const dedupState = createMessageDedupState()
+  const fallbackOnlyMessageIndexes = new Map<string, number>()
+  const mergedMessages: ChatLabMergedMessage[] = []
+
+  for (const { source, meta, sourceIndex } of metas) {
     const platform = meta?.platform || 'unknown'
+    const messageScope = scopeResolver.resolve(sourceIndex)
     source.streamMessages(10000, (messages) => {
       for (const msg of messages) {
         const nid = normalizePlatformId(msg.senderPlatformId, platform, collidingIds)
-        const key = generateMessageKey(msg.timestamp, nid, msg.content ?? null)
-        if (seenKeys.has(key)) continue
-        seenKeys.add(key)
+        const platformMessageId = normalizeMessageId(msg.platformMessageId, messageScope)
+        const replyToMessageId = normalizeMessageId(msg.replyToMessageId, messageScope)
+        const dedupMessage = {
+          platformMessageId,
+          timestamp: msg.timestamp,
+          senderPlatformId: nid,
+          type: msg.type,
+          content: msg.content ?? null,
+          replyToMessageId,
+        }
+        const fallbackKey = generateFallbackMessageKey(dedupMessage)
+        const shouldBackfillId = Boolean(
+          platformMessageId &&
+          !dedupState.platformMessageIds.has(platformMessageId) &&
+          dedupState.fallbackOnlyKeys.has(fallbackKey)
+        )
+
+        if (registerMessageAndCheckDuplicate(dedupMessage, dedupState)) {
+          if (shouldBackfillId) {
+            const retainedIndex = fallbackOnlyMessageIndexes.get(fallbackKey)
+            if (retainedIndex !== undefined) {
+              mergedMessages[retainedIndex].platformMessageId = platformMessageId
+              fallbackOnlyMessageIndexes.delete(fallbackKey)
+            }
+          }
+          continue
+        }
+
+        const retainedIndex = mergedMessages.length
         mergedMessages.push({
+          platformMessageId,
           sender: nid,
           accountName: msg.senderAccountName,
           groupNickname: msg.senderGroupNickname,
           timestamp: msg.timestamp,
           type: msg.type,
           content: msg.content,
+          replyToMessageId,
         })
+        if (!platformMessageId) fallbackOnlyMessageIndexes.set(fallbackKey, retainedIndex)
       }
     })
   }
@@ -162,7 +381,6 @@ export function buildMergedOutput(
     messageCount: source.getMessageCount(),
   }))
 
-  const uniquePlatforms = [...new Set(metas.map(({ meta }) => meta?.platform || 'unknown'))]
   const platform = uniquePlatforms.length === 1 ? uniquePlatforms[0] : 'mixed'
 
   const groupIds = new Set(metas.map(({ meta }) => meta?.groupId).filter(Boolean))

packages/node-runtime/src/merger/temp-db.ts

@@ -12,8 +12,8 @@ import * as fs from 'fs'
 import * as path from 'path'
 import type { DatabaseAdapter } from '@openchatlab/core'
 import { CHATLAB_FORMAT_VERSION, type ParsedMember, type ParsedMessage } from '@openchatlab/shared-types'
-import type { MergerDataSource, MergerSourceMeta } from './index'
-import type { MergerMember, MergerMessage } from '@openchatlab/core'
+import type { MergerDataSource, MergerInputMessage, MergerSourceMeta } from './index'
+import type { MergerMember } from '@openchatlab/core'
 
 // ==================== Temp DB schema (simplified for merge preview) ====================
 
@@ -36,12 +36,14 @@ export const TEMP_DB_SCHEMA = `
 
   CREATE TABLE IF NOT EXISTS message (
     id INTEGER PRIMARY KEY AUTOINCREMENT,
+    platform_message_id TEXT,
     sender_platform_id TEXT NOT NULL,
     sender_account_name TEXT,
     sender_group_nickname TEXT,
     timestamp INTEGER NOT NULL,
     type INTEGER NOT NULL,
-    content TEXT
+    content TEXT,
+    reply_to_message_id TEXT
   );
 
   CREATE INDEX IF NOT EXISTS idx_message_ts ON message(timestamp);
@@ -88,8 +90,10 @@ export class TempDbWriter {
 
   writeMessages(messages: ParsedMessage[]): void {
     const insert = this.db.prepare(
-      `INSERT INTO message (sender_platform_id, sender_account_name, sender_group_nickname, timestamp, type, content)
-       VALUES (?, ?, ?, ?, ?, ?)`
+      `INSERT INTO message (
+         platform_message_id, sender_platform_id, sender_account_name, sender_group_nickname,
+         timestamp, type, content, reply_to_message_id
+       ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
     )
     const memberInsert = this.db.prepare(
       'INSERT OR IGNORE INTO member (platform_id, account_name, group_nickname, avatar) VALUES (?, ?, ?, ?)'
@@ -100,12 +104,14 @@ export class TempDbWriter {
         memberInsert.run(msg.senderPlatformId, msg.senderAccountName || null, msg.senderGroupNickname || null, null)
       }
       insert.run(
+        msg.platformMessageId || null,
         msg.senderPlatformId,
         msg.senderAccountName || null,
         msg.senderGroupNickname || null,
         msg.timestamp,
         msg.type,
-        msg.content || null
+        msg.content || null,
+        msg.replyToMessageId || null
       )
       this.messageCount++
     }
@@ -181,30 +187,35 @@ export class TempDbReader {
 
   streamMessages(batchSize: number, callback: (messages: ParsedMessage[]) => void): void {
     const stmt = this.db.prepare(`
-      SELECT sender_platform_id, sender_account_name, sender_group_nickname, timestamp, type, content
-      FROM message ORDER BY timestamp ASC LIMIT ? OFFSET ?
+      SELECT platform_message_id, sender_platform_id, sender_account_name, sender_group_nickname,
+             timestamp, type, content, reply_to_message_id
+      FROM message ORDER BY timestamp ASC, id ASC LIMIT ? OFFSET ?
     `)
 
     let offset = 0
     while (true) {
       const rows = stmt.all(batchSize, offset) as Array<{
+        platform_message_id: string | null
         sender_platform_id: string
         sender_account_name: string | null
         sender_group_nickname: string | null
         timestamp: number
         type: number
         content: string | null
+        reply_to_message_id: string | null
       }>
 
       if (rows.length === 0) break
 
       const messages: ParsedMessage[] = rows.map((r) => ({
+        platformMessageId: r.platform_message_id || undefined,
         senderPlatformId: r.sender_platform_id,
         senderAccountName: r.sender_account_name || r.sender_platform_id,
         senderGroupNickname: r.sender_group_nickname || undefined,
         timestamp: r.timestamp,
         type: r.type as ParsedMessage['type'],
         content: r.content,
+        replyToMessageId: r.reply_to_message_id || undefined,
       }))
 
       callback(messages)
@@ -215,25 +226,30 @@ export class TempDbReader {
   getAllMessages(): ParsedMessage[] {
     const rows = this.db
       .prepare(
-        `SELECT sender_platform_id, sender_account_name, sender_group_nickname, timestamp, type, content
-         FROM message ORDER BY timestamp ASC`
+        `SELECT platform_message_id, sender_platform_id, sender_account_name, sender_group_nickname,
+                timestamp, type, content, reply_to_message_id
+         FROM message ORDER BY timestamp ASC, id ASC`
       )
       .all() as Array<{
+      platform_message_id: string | null
       sender_platform_id: string
       sender_account_name: string | null
       sender_group_nickname: string | null
       timestamp: number
       type: number
       content: string | null
+      reply_to_message_id: string | null
     }>
 
     return rows.map((r) => ({
+      platformMessageId: r.platform_message_id || undefined,
       senderPlatformId: r.sender_platform_id,
       senderAccountName: r.sender_account_name || r.sender_platform_id,
       senderGroupNickname: r.sender_group_nickname || undefined,
       timestamp: r.timestamp,
       type: r.type as ParsedMessage['type'],
       content: r.content,
+      replyToMessageId: r.reply_to_message_id || undefined,
     }))
   }
 
@@ -271,16 +287,18 @@ function createDataSourceFromReader(reader: TempDbReader): MergerDataSource {
     getMessageCount(): number {
       return reader.getMessageCount()
     },
-    streamMessages(batchSize: number, callback: (messages: MergerMessage[]) => void): void {
+    streamMessages(batchSize: number, callback: (messages: MergerInputMessage[]) => void): void {
       reader.streamMessages(batchSize, (messages) => {
         callback(
           messages.map((msg) => ({
             senderPlatformId: msg.senderPlatformId,
+            platformMessageId: msg.platformMessageId,
             senderAccountName: msg.senderAccountName,
             senderGroupNickname: msg.senderGroupNickname,
             timestamp: msg.timestamp,
             type: msg.type,
             content: msg.content,
+            replyToMessageId: msg.replyToMessageId,
           }))
         )
       })
@@ -295,12 +313,14 @@ export interface ExportedSession {
   meta: { name: string; platform: string; type: string; groupId?: string; groupAvatar?: string }
   members: Array<{ platformId: string; accountName: string; groupNickname?: string; avatar?: string }>
   messages: Array<{
+    platformMessageId?: string
     sender: string
     accountName: string
     groupNickname?: string
     timestamp: number
     type: number
     content: string | null
+    replyToMessageId?: string
   }>
 }
 
@@ -325,16 +345,19 @@ export function exportSessionToJson(db: DatabaseAdapter): ExportedSession {
   const messages = db
     .prepare(
       `SELECT m.platform_id as sender, msg.sender_account_name as accountName,
-              msg.sender_group_nickname as groupNickname, msg.ts as timestamp, msg.type, msg.content
-       FROM message msg JOIN member m ON msg.sender_id = m.id ORDER BY msg.ts`
+              msg.sender_group_nickname as groupNickname, msg.ts as timestamp, msg.type, msg.content,
+              msg.platform_message_id as platformMessageId, msg.reply_to_message_id as replyToMessageId
+       FROM message msg JOIN member m ON msg.sender_id = m.id ORDER BY msg.ts, msg.id`
     )
     .all() as Array<{
+    platformMessageId?: string
     sender: string
     accountName?: string
     groupNickname?: string
     timestamp: number
     type: number
     content?: string
+    replyToMessageId?: string
   }>
 
   return {
@@ -358,12 +381,14 @@ export function exportSessionToJson(db: DatabaseAdapter): ExportedSession {
       avatar: m.avatar,
     })),
     messages: messages.map((msg) => ({
+      platformMessageId: msg.platformMessageId || undefined,
       sender: msg.sender,
       accountName: msg.accountName || msg.sender,
       groupNickname: msg.groupNickname || undefined,
       timestamp: msg.timestamp,
       type: msg.type,
       content: msg.content ?? null,
+      replyToMessageId: msg.replyToMessageId || undefined,
     })),
   }
 }