fix(import): stabilize QQ automatic session matching

Match ownerless QQ exports safely across UID and UIN drift, preserve merged message scopes through single and batch imports, and apply only sender mappings proven by overlapping platform message IDs. Reject incomplete group mappings to avoid member identity corruption.

apps/cli/src/import/stream-import-flow.test.ts

@@ -4,7 +4,7 @@ import os from 'node:os'
 import path from 'node:path'
 import test from 'node:test'
 import Database from 'better-sqlite3'
-import type { PathProvider } from '@openchatlab/core'
+import { CHAT_DB_SCHEMA, type PathProvider } from '@openchatlab/core'
 import {
   DatabaseManager,
   IMPORT_IN_PROGRESS_ERROR_KEY,
@@ -81,6 +81,58 @@ function writeBatchChatFile(dir: string, filename: string, groupId: string, cont
   return filePath
 }
 
+function writeScopedBatchUpdate(dir: string): string {
+  const filePath = path.join(dir, 'scoped-update.json')
+  fs.writeFileSync(
+    filePath,
+    JSON.stringify({
+      chatlab: { version: '0.0.2', exportedAt: 1783840200 },
+      meta: { name: 'Private chat', platform: 'qq', type: 'private' },
+      members: [
+        { platformId: '10001', accountName: 'Owner' },
+        { platformId: '20002', accountName: 'Peer' },
+      ],
+      messages: Array.from({ length: 6 }, (_, index) => ({
+        sender: index % 2 === 0 ? '20002' : '10001',
+        accountName: index % 2 === 0 ? 'Peer' : 'Owner',
+        timestamp: 1783840200 + index,
+        type: 0,
+        content: `message ${index}`,
+        platformMessageId: `message-${index}`,
+      })),
+    })
+  )
+  return filePath
+}
+
+function seedScopedBatchSession(root: string): void {
+  const db = new Database(path.join(root, 'data', 'databases', 'scoped-target.db'), { nativeBinding })
+  db.exec(CHAT_DB_SCHEMA)
+  db.prepare(
+    `INSERT INTO meta (name, platform, type, imported_at, schema_version)
+     VALUES (?, ?, ?, ?, ?)`
+  ).run('Private chat', 'qq', 'private', 1783840200, 6)
+  const insertMember = db.prepare('INSERT INTO member (platform_id, account_name) VALUES (?, ?)')
+  const peerId = Number(insertMember.run('uid-peer', 'Peer').lastInsertRowid)
+  const ownerId = Number(insertMember.run('uid-owner', 'Owner').lastInsertRowid)
+  const insertMessage = db.prepare(
+    `INSERT INTO message (sender_id, sender_account_name, ts, type, content, platform_message_id)
+     VALUES (?, ?, ?, ?, ?, ?)`
+  )
+  for (let index = 0; index < 5; index++) {
+    const isPeer = index % 2 === 0
+    insertMessage.run(
+      isPeer ? peerId : ownerId,
+      isPeer ? 'Peer' : 'Owner',
+      1783840200 + index,
+      0,
+      `message ${index}`,
+      `__chatlab_message_scope__0__message-${index}`
+    )
+  }
+  db.close()
+}
+
 test('streamImport raises the data directory gate after creating a current-schema database', async () => {
   const root = makeTempDir()
   fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true })
@@ -233,6 +285,32 @@ test('autoImportBatch holds one lock, coalesces the same target, and preserves i
   assert.deepEqual(row, { messages: 2, gapThreshold: 7200 })
 })
 
+test('autoImportBatch forwards the matched merger scope into incremental deduplication', async (t) => {
+  const root = makeTempDir()
+  t.after(() => fs.rmSync(root, { recursive: true, force: true }))
+  fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true })
+  seedScopedBatchSession(root)
+  const manager = new DatabaseManager(createPathProvider(root), {
+    nativeBinding,
+    runtime: { version: '0.25.1', kind: 'cli' },
+  })
+
+  const results = await autoImportBatch(manager, [{ id: 'scoped-update', filePath: writeScopedBatchUpdate(root) }])
+
+  assert.equal(results[0].status, 'success')
+  assert.equal(results[0].status === 'success' && results[0].result.importMode, 'incremental')
+  assert.equal(results[0].status === 'success' && results[0].result.newMessageCount, 1)
+  assert.equal(results[0].status === 'success' && results[0].result.duplicateCount, 5)
+
+  const db = manager.openRawSessionDatabase('scoped-target', { readonly: true })
+  const rows = db.prepare('SELECT platform_message_id FROM message ORDER BY ts, id').all() as Array<{
+    platform_message_id: string
+  }>
+  db.close()
+  assert.equal(rows.length, 6)
+  assert.equal(rows.at(-1)?.platform_message_id, '__chatlab_message_scope__0__message-5')
+})
+
 test('autoImportBatch rejects an external writer and cancels work that has not started', async (t) => {
   const root = makeTempDir()
   t.after(() => fs.rmSync(root, { recursive: true, force: true }))

apps/cli/src/import/stream-import.ts

@@ -244,7 +244,7 @@ async function autoImportUnlocked(
           updateCompatibilityGate,
           itemProgress
         ),
-      appendSession: (targetSessionId, sourcePath, sourceFormatOptions, itemProgress) =>
+      appendSession: (targetSessionId, sourcePath, sourceFormatOptions, itemProgress, context) =>
         incrementalImportUnlocked(
           dbManager,
           targetSessionId,
@@ -252,6 +252,8 @@ async function autoImportUnlocked(
           {
             ...sourceFormatOptions,
             onProgress: itemProgress ?? progressAdapter,
+            platformMessageIdScope: context?.platformMessageIdScope,
+            senderPlatformIdMappings: context?.senderPlatformIdMappings,
           },
           updateCompatibilityGate
         ),
@@ -316,7 +318,7 @@ export async function autoImportBatch(
               false,
               itemProgress
             ),
-          appendSession: (targetSessionId, sourcePath, sourceFormatOptions, itemProgress) =>
+          appendSession: (targetSessionId, sourcePath, sourceFormatOptions, itemProgress, context) =>
             incrementalImportUnlocked(
               dbManager,
               targetSessionId,
@@ -324,6 +326,8 @@ export async function autoImportBatch(
               {
                 ...sourceFormatOptions,
                 onProgress: itemProgress,
+                platformMessageIdScope: context?.platformMessageIdScope,
+                senderPlatformIdMappings: context?.senderPlatformIdMappings,
               },
               false
             ),
@@ -399,10 +403,12 @@ export async function analyzeAutoImport(
           formatId: typeof sourceFormatOptions?.formatId === 'string' ? sourceFormatOptions.formatId : undefined,
           chatIndex: typeof sourceFormatOptions?.chatIndex === 'number' ? sourceFormatOptions.chatIndex : undefined,
         }),
-      analyzeAppendSession: (targetSessionId, sourcePath, sourceFormatOptions) =>
+      analyzeAppendSession: (targetSessionId, sourcePath, sourceFormatOptions, context) =>
         sharedAnalyzeIncremental(targetSessionId, sourcePath, buildIncrementalDeps(dbManager, progressAdapter), {
           formatId: typeof sourceFormatOptions?.formatId === 'string' ? sourceFormatOptions.formatId : undefined,
           chatIndex: typeof sourceFormatOptions?.chatIndex === 'number' ? sourceFormatOptions.chatIndex : undefined,
+          platformMessageIdScope: context?.platformMessageIdScope,
+          senderPlatformIdMappings: context?.senderPlatformIdMappings,
         }),
     },
     {