fix(import): harden push and pull import guards

Combine the PR review fixes for push import validation, deduplication, system sender handling, and pull import lock scoping into one reviewable change.

apps/desktop/main/api/routes/import.ts

@@ -110,7 +110,10 @@ function idempotencyFail(key: string | undefined): void {
   idempotencyCache.delete(key)
 }
 
-export function getImportingStatus(): boolean {
+export function getImportingStatus(sessionId?: string): boolean {
+  if (sessionId) {
+    return isImporting.has(sessionId) || isImporting.has('__legacy__')
+  }
   return isImporting.size > 0
 }
 

packages/node-runtime/src/services/push-importer.test.ts

@@ -0,0 +1,511 @@
+import assert from 'node:assert/strict'
+import fs from 'node:fs'
+import os from 'node:os'
+import path from 'node:path'
+import test from 'node:test'
+import { DatabaseManager } from '../database-manager'
+import type { PushImportPayload } from './push-importer'
+import { pushImport } from './push-importer'
+
+const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node')
+
+function makeTempDir(): string {
+  const baseDir = fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir()
+  return fs.mkdtempSync(path.join(baseDir, 'chatlab-push-import-'))
+}
+
+function createDatabaseManager(rootDir: string): DatabaseManager {
+  return new DatabaseManager(
+    {
+      getSystemDir: () => rootDir,
+      getUserDataDir: () => rootDir,
+      getDatabaseDir: () => path.join(rootDir, 'databases'),
+      getVectorDir: () => path.join(rootDir, 'vector'),
+      getAiDataDir: () => path.join(rootDir, 'ai'),
+      getSettingsDir: () => path.join(rootDir, 'settings'),
+      getCacheDir: () => path.join(rootDir, 'cache'),
+      getTempDir: () => path.join(rootDir, 'temp'),
+      getLogsDir: () => path.join(rootDir, 'logs'),
+      getDownloadsDir: () => rootDir,
+    },
+    { nativeBinding, allowMissingRuntimeForTests: true }
+  )
+}
+
+test('deduplicates initial push batch when a platform-id message is repeated without platform id', async (t) => {
+  const tempDir = makeTempDir()
+  t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
+
+  const manager = createDatabaseManager(tempDir)
+  const payload: PushImportPayload = {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Push Import Session', platform: 'wechat', type: 'private' },
+    members: [{ platformId: 'wxid_alice', accountName: 'Alice' }],
+    messages: [
+      {
+        platformMessageId: 'msg-1',
+        sender: 'wxid_alice',
+        accountName: 'Alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'same content',
+      },
+      {
+        sender: 'wxid_alice',
+        accountName: 'Alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'same content',
+      },
+    ],
+  }
+
+  const outcome = await pushImport(manager, 'initial-dedup', payload)
+  assert.equal(outcome.ok, true)
+  if (!outcome.ok) return
+  assert.deepEqual(outcome.result.batch, {
+    receivedCount: 2,
+    writtenCount: 1,
+    duplicateCount: 1,
+  })
+
+  const db = manager.openRawSessionDatabase('initial-dedup', { readonly: true })
+  try {
+    const row = db.prepare('SELECT COUNT(*) as count FROM message').get() as { count: number }
+    assert.equal(row.count, 1)
+  } finally {
+    db.close()
+  }
+})
+
+test('deduplicates initial push batch when a no-platform-id message is repeated with platform id', async (t) => {
+  const tempDir = makeTempDir()
+  t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
+
+  const manager = createDatabaseManager(tempDir)
+  const payload: PushImportPayload = {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Push Import Session', platform: 'wechat', type: 'private' },
+    members: [{ platformId: 'wxid_alice', accountName: 'Alice' }],
+    messages: [
+      {
+        sender: 'wxid_alice',
+        accountName: 'Alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'same content',
+      },
+      {
+        platformMessageId: 'msg-1',
+        sender: 'wxid_alice',
+        accountName: 'Alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'same content',
+      },
+    ],
+  }
+
+  const outcome = await pushImport(manager, 'initial-reverse-dedup', payload)
+  assert.equal(outcome.ok, true)
+  if (!outcome.ok) return
+  assert.deepEqual(outcome.result.batch, {
+    receivedCount: 2,
+    writtenCount: 1,
+    duplicateCount: 1,
+  })
+
+  const db = manager.openRawSessionDatabase('initial-reverse-dedup', { readonly: true })
+  try {
+    const row = db.prepare('SELECT COUNT(*) as count FROM message').get() as { count: number }
+    assert.equal(row.count, 1)
+  } finally {
+    db.close()
+  }
+})
+
+test('deduplicates incremental push when a later platform-id message matches an existing content hash', async (t) => {
+  const tempDir = makeTempDir()
+  t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
+
+  const manager = createDatabaseManager(tempDir)
+  const initialPayload: PushImportPayload = {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Push Import Session', platform: 'wechat', type: 'private' },
+    members: [{ platformId: 'wxid_alice', accountName: 'Alice' }],
+    messages: [
+      {
+        sender: 'wxid_alice',
+        accountName: 'Alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'same content',
+      },
+    ],
+  }
+
+  const initialOutcome = await pushImport(manager, 'incremental-pmid-hash-dedup', initialPayload)
+  assert.equal(initialOutcome.ok, true)
+
+  const duplicatePayload: PushImportPayload = {
+    messages: [
+      {
+        platformMessageId: 'msg-1',
+        sender: 'wxid_alice',
+        accountName: 'Alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'same content',
+      },
+    ],
+  }
+
+  const outcome = await pushImport(manager, 'incremental-pmid-hash-dedup', duplicatePayload)
+  assert.equal(outcome.ok, true)
+  if (!outcome.ok) return
+  assert.deepEqual(outcome.result.batch, {
+    receivedCount: 1,
+    writtenCount: 0,
+    duplicateCount: 1,
+  })
+
+  const db = manager.openRawSessionDatabase('incremental-pmid-hash-dedup', { readonly: true })
+  try {
+    const row = db.prepare('SELECT COUNT(*) as count FROM message').get() as { count: number }
+    assert.equal(row.count, 1)
+  } finally {
+    db.close()
+  }
+})
+
+test('rejects non-array messages before applying incremental metadata updates', async (t) => {
+  const tempDir = makeTempDir()
+  t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
+
+  const manager = createDatabaseManager(tempDir)
+  const initialPayload: PushImportPayload = {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Original Session', platform: 'wechat', type: 'private' },
+    members: [{ platformId: 'wxid_alice', accountName: 'Alice' }],
+    messages: [
+      {
+        sender: 'wxid_alice',
+        accountName: 'Alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'hello',
+      },
+    ],
+  }
+
+  const initialOutcome = await pushImport(manager, 'invalid-messages', initialPayload)
+  assert.equal(initialOutcome.ok, true)
+
+  const malformedPayload = {
+    meta: { name: 'Renamed Session', platform: 'wechat', type: 'private' },
+    messages: {},
+  } as unknown as PushImportPayload
+
+  const outcome = await pushImport(manager, 'invalid-messages', malformedPayload)
+  assert.equal(outcome.ok, false)
+  if (outcome.ok) return
+  assert.equal(outcome.reason, 'invalid_payload')
+
+  const db = manager.openRawSessionDatabase('invalid-messages', { readonly: true })
+  try {
+    const row = db.prepare('SELECT name FROM meta').get() as { name: string }
+    assert.equal(row.name, 'Original Session')
+  } finally {
+    db.close()
+  }
+})
+
+test('rejects non-array members before applying incremental metadata updates', async (t) => {
+  const tempDir = makeTempDir()
+  t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
+
+  const manager = createDatabaseManager(tempDir)
+  const initialPayload: PushImportPayload = {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Original Session', platform: 'wechat', type: 'private' },
+    members: [{ platformId: 'wxid_alice', accountName: 'Alice' }],
+    messages: [
+      {
+        sender: 'wxid_alice',
+        accountName: 'Alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'hello',
+      },
+    ],
+  }
+
+  const initialOutcome = await pushImport(manager, 'invalid-members', initialPayload)
+  assert.equal(initialOutcome.ok, true)
+
+  const malformedPayload = {
+    meta: { name: 'Renamed Session', platform: 'wechat', type: 'private' },
+    members: {},
+    messages: [
+      {
+        sender: 'wxid_alice',
+        timestamp: 1780330833,
+        type: 0,
+        content: 'still valid',
+      },
+    ],
+  } as unknown as PushImportPayload
+
+  const outcome = await pushImport(manager, 'invalid-members', malformedPayload)
+  assert.equal(outcome.ok, false)
+  if (outcome.ok) return
+  assert.equal(outcome.reason, 'invalid_payload')
+
+  const db = manager.openRawSessionDatabase('invalid-members', { readonly: true })
+  try {
+    const row = db.prepare('SELECT name FROM meta').get() as { name: string }
+    assert.equal(row.name, 'Original Session')
+  } finally {
+    db.close()
+  }
+})
+
+test('rejects non-string member platform IDs before applying incremental metadata updates', async (t) => {
+  const tempDir = makeTempDir()
+  t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
+
+  const manager = createDatabaseManager(tempDir)
+  const initialPayload: PushImportPayload = {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Original Session', platform: 'wechat', type: 'private' },
+    members: [{ platformId: 'wxid_alice', accountName: 'Alice' }],
+    messages: [
+      {
+        sender: 'wxid_alice',
+        accountName: 'Alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'hello',
+      },
+    ],
+  }
+
+  const initialOutcome = await pushImport(manager, 'invalid-member-platform-id', initialPayload)
+  assert.equal(initialOutcome.ok, true)
+
+  const malformedPayload = {
+    meta: { name: 'Renamed Session', platform: 'wechat', type: 'private' },
+    members: [{ platformId: { id: 'wxid_alice' }, accountName: 'Alice' }],
+    messages: [
+      {
+        sender: 'wxid_alice',
+        timestamp: 1780330833,
+        type: 0,
+        content: 'still valid',
+      },
+    ],
+  } as unknown as PushImportPayload
+
+  const outcome = await pushImport(manager, 'invalid-member-platform-id', malformedPayload)
+  assert.equal(outcome.ok, false)
+  if (outcome.ok) return
+  assert.equal(outcome.reason, 'invalid_payload')
+
+  const db = manager.openRawSessionDatabase('invalid-member-platform-id', { readonly: true })
+  try {
+    const row = db.prepare('SELECT name FROM meta').get() as { name: string }
+    assert.equal(row.name, 'Original Session')
+  } finally {
+    db.close()
+  }
+})
+
+test('rejects non-string message senders before applying incremental metadata updates', async (t) => {
+  const tempDir = makeTempDir()
+  t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
+
+  const manager = createDatabaseManager(tempDir)
+  const initialPayload: PushImportPayload = {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Original Session', platform: 'wechat', type: 'private' },
+    members: [{ platformId: 'wxid_alice', accountName: 'Alice' }],
+    messages: [
+      {
+        sender: 'wxid_alice',
+        accountName: 'Alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'hello',
+      },
+    ],
+  }
+
+  const initialOutcome = await pushImport(manager, 'invalid-sender', initialPayload)
+  assert.equal(initialOutcome.ok, true)
+
+  const malformedPayload = {
+    meta: { name: 'Renamed Session', platform: 'wechat', type: 'private' },
+    messages: [
+      {
+        sender: { id: 'wxid_alice' },
+        timestamp: 1780330833,
+        type: 0,
+        content: 'still valid',
+      },
+    ],
+  } as unknown as PushImportPayload
+
+  const outcome = await pushImport(manager, 'invalid-sender', malformedPayload)
+  assert.equal(outcome.ok, false)
+  if (outcome.ok) return
+  assert.equal(outcome.reason, 'invalid_payload')
+
+  const db = manager.openRawSessionDatabase('invalid-sender', { readonly: true })
+  try {
+    const row = db.prepare('SELECT name FROM meta').get() as { name: string }
+    assert.equal(row.name, 'Original Session')
+  } finally {
+    db.close()
+  }
+})
+
+test('rejects non-string reply targets before applying incremental metadata updates', async (t) => {
+  const tempDir = makeTempDir()
+  t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
+
+  const manager = createDatabaseManager(tempDir)
+  const initialPayload: PushImportPayload = {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Original Session', platform: 'wechat', type: 'private' },
+    members: [{ platformId: 'wxid_alice', accountName: 'Alice' }],
+    messages: [
+      {
+        sender: 'wxid_alice',
+        accountName: 'Alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'hello',
+      },
+    ],
+  }
+
+  const initialOutcome = await pushImport(manager, 'invalid-reply-target', initialPayload)
+  assert.equal(initialOutcome.ok, true)
+
+  const malformedPayload = {
+    meta: { name: 'Renamed Session', platform: 'wechat', type: 'private' },
+    messages: [
+      {
+        sender: 'wxid_alice',
+        timestamp: 1780330833,
+        type: 0,
+        content: 'still valid',
+        replyToMessageId: { id: 'msg-1' },
+      },
+    ],
+  } as unknown as PushImportPayload
+
+  const outcome = await pushImport(manager, 'invalid-reply-target', malformedPayload)
+  assert.equal(outcome.ok, false)
+  if (outcome.ok) return
+  assert.equal(outcome.reason, 'invalid_payload')
+
+  const db = manager.openRawSessionDatabase('invalid-reply-target', { readonly: true })
+  try {
+    const row = db.prepare('SELECT name FROM meta').get() as { name: string }
+    assert.equal(row.name, 'Original Session')
+  } finally {
+    db.close()
+  }
+})
+
+test('treats auto-created SYSTEM sender as a system member during initial import', async (t) => {
+  const tempDir = makeTempDir()
+  t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
+
+  const manager = createDatabaseManager(tempDir)
+  const payload: PushImportPayload = {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'System Session', platform: 'wechat', type: 'group' },
+    messages: [
+      {
+        sender: 'SYSTEM',
+        timestamp: 1780330832,
+        type: 80,
+        content: 'Alice joined the group',
+      },
+    ],
+  }
+
+  const outcome = await pushImport(manager, 'system-sender', payload)
+  assert.equal(outcome.ok, true)
+  if (!outcome.ok) return
+  assert.equal(outcome.result.session.memberCount, 0)
+
+  const db = manager.openRawSessionDatabase('system-sender', { readonly: true })
+  try {
+    const member = db.prepare('SELECT platform_id, account_name FROM member').get() as {
+      platform_id: string
+      account_name: string
+    }
+    assert.deepEqual(member, {
+      platform_id: 'SYSTEM',
+      account_name: '系统消息',
+    })
+  } finally {
+    db.close()
+  }
+})
+
+test('keeps SYSTEM member canonical during incremental member upserts', async (t) => {
+  const tempDir = makeTempDir()
+  t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
+
+  const manager = createDatabaseManager(tempDir)
+  const initialPayload: PushImportPayload = {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'System Session', platform: 'wechat', type: 'group' },
+    messages: [
+      {
+        sender: 'SYSTEM',
+        timestamp: 1780330832,
+        type: 80,
+        content: 'Alice joined the group',
+      },
+    ],
+  }
+
+  const initialOutcome = await pushImport(manager, 'system-member-upsert', initialPayload)
+  assert.equal(initialOutcome.ok, true)
+
+  const incrementalPayload: PushImportPayload = {
+    members: [{ platformId: 'SYSTEM', accountName: 'Bot' }],
+    messages: [
+      {
+        sender: 'SYSTEM',
+        timestamp: 1780330833,
+        type: 80,
+        content: 'Bob joined the group',
+      },
+    ],
+  }
+
+  const outcome = await pushImport(manager, 'system-member-upsert', incrementalPayload)
+  assert.equal(outcome.ok, true)
+  if (!outcome.ok) return
+  assert.equal(outcome.result.session.memberCount, 0)
+
+  const db = manager.openRawSessionDatabase('system-member-upsert', { readonly: true })
+  try {
+    const member = db.prepare('SELECT platform_id, account_name FROM member').get() as {
+      platform_id: string
+      account_name: string
+    }
+    assert.deepEqual(member, {
+      platform_id: 'SYSTEM',
+      account_name: '系统消息',
+    })
+  } finally {
+    db.close()
+  }
+})

packages/node-runtime/src/services/push-importer.ts

@@ -20,6 +20,8 @@ import { writeParseResultToDb } from '../import'
 
 // Per-session lock: concurrent imports to different sessions are fine (separate DB files).
 const importInProgress = new Set<string>()
+const SYSTEM_SENDER_ID = 'SYSTEM'
+const SYSTEM_MEMBER_NAME = '系统消息'
 
 export interface PushImportMessage {
   sender: string
@@ -74,7 +76,7 @@ export type PushImportOutcome =
 
 function validatePayload(payload: PushImportPayload, isNew: boolean): string | null {
   const messages = payload.messages
-  if (!messages || messages.length === 0) return 'messages is required and must contain at least one message'
+  if (!Array.isArray(messages) || messages.length === 0) return 'messages is required and must be a non-empty array'
 
   if (isNew) {
     if (!payload.chatlab) return 'chatlab is required for new sessions'
@@ -93,15 +95,17 @@ function validatePayload(payload: PushImportPayload, isNew: boolean): string | n
     return `options.memberUpdateMode must be 'upsert' or 'none'`
   }
 
+  if (payload.members !== undefined && !Array.isArray(payload.members)) return 'members must be an array'
   if (payload.members) {
     for (let i = 0; i < payload.members.length; i++) {
-      if (!payload.members[i].platformId) return `members[${i}].platformId is required`
+      if (typeof payload.members[i].platformId !== 'string' || payload.members[i].platformId.length === 0)
+        return `members[${i}].platformId must be a string`
     }
   }
 
   for (let i = 0; i < messages.length; i++) {
     const msg = messages[i]
-    if (!msg.sender) return `messages[${i}].sender is required`
+    if (typeof msg.sender !== 'string' || msg.sender.length === 0) return `messages[${i}].sender must be a string`
     if (typeof msg.timestamp !== 'number' || msg.timestamp <= 0)
       return `messages[${i}].timestamp must be a positive number`
     if (typeof msg.type !== 'number') return `messages[${i}].type must be a number`
@@ -109,6 +113,8 @@ function validatePayload(payload: PushImportPayload, isNew: boolean): string | n
       return `messages[${i}].content must be a string or null`
     if (msg.platformMessageId !== undefined && typeof msg.platformMessageId !== 'string')
       return `messages[${i}].platformMessageId must be a string`
+    if (msg.replyToMessageId !== undefined && typeof msg.replyToMessageId !== 'string')
+      return `messages[${i}].replyToMessageId must be a string`
   }
 
   return null
@@ -120,10 +126,20 @@ function queryStats(db: DatabaseAdapter): PushImportResult['session'] {
     first: number | null
     last: number | null
   }
-  const memberRow = db.prepare('SELECT COUNT(*) as cnt FROM member').get() as { cnt: number }
+  const memberRow = db
+    .prepare(`SELECT COUNT(*) as cnt FROM member WHERE COALESCE(account_name, '') != ?`)
+    .get(SYSTEM_MEMBER_NAME) as { cnt: number }
   return { totalCount: row.total, memberCount: memberRow.cnt, firstTimestamp: row.first, lastTimestamp: row.last }
 }
 
+function normalizeSenderAccountName(sender: string, accountName: string | undefined): string | undefined {
+  return sender === SYSTEM_SENDER_ID ? SYSTEM_MEMBER_NAME : accountName
+}
+
+function normalizeSenderGroupNickname(sender: string, groupNickname: string | undefined): string | undefined {
+  return sender === SYSTEM_SENDER_ID ? SYSTEM_MEMBER_NAME : groupNickname
+}
+
 function writeMessages(
   db: DatabaseAdapter,
   messages: PushImportMessage[],
@@ -152,17 +168,21 @@ function writeMessages(
 
   db.transaction(() => {
     for (const msg of sorted) {
+      const key = generateMessageKey(msg.timestamp, msg.sender, msg.content ?? null)
       if (msg.platformMessageId) {
         if (existingPmids.has(msg.platformMessageId)) {
           duplicateCount++
           continue
         }
+        if (existingKeys.has(key)) {
+          duplicateCount++
+          continue
+        }
         existingPmids.add(msg.platformMessageId)
         // Also register the content hash so a same-content no-pmid copy later
         // in this batch is caught by the fallback dedup path.
-        existingKeys.add(generateMessageKey(msg.timestamp, msg.sender, msg.content ?? null))
+        existingKeys.add(key)
       } else {
-        const key = generateMessageKey(msg.timestamp, msg.sender, msg.content ?? null)
         if (existingKeys.has(key)) {
           duplicateCount++
           continue
@@ -172,7 +192,7 @@ function writeMessages(
 
       let memberId = memberIdCache.get(msg.sender)
       if (!memberId) {
-        insertMinimalMember.run(msg.sender, msg.accountName || null)
+        insertMinimalMember.run(msg.sender, normalizeSenderAccountName(msg.sender, msg.accountName) || null)
         const row = getMemberId.get(msg.sender) as { id: number } | undefined
         if (row) {
           memberId = row.id
@@ -183,8 +203,8 @@ function writeMessages(
 
       const result = insertMsg.run(
         memberId,
-        msg.accountName || null,
-        msg.groupNickname || null,
+        normalizeSenderAccountName(msg.sender, msg.accountName) || null,
+        normalizeSenderGroupNickname(msg.sender, msg.groupNickname) || null,
         msg.timestamp,
         msg.type,
         msg.content ?? null,
@@ -214,7 +234,11 @@ function fullImport(
   const extraMembers: PushImportMember[] = []
   for (const msg of messages) {
     if (!knownIds.has(msg.sender)) {
-      extraMembers.push({ platformId: msg.sender, accountName: msg.accountName })
+      extraMembers.push({
+        platformId: msg.sender,
+        accountName: normalizeSenderAccountName(msg.sender, msg.accountName),
+        groupNickname: normalizeSenderGroupNickname(msg.sender, msg.groupNickname),
+      })
       knownIds.add(msg.sender)
     }
   }
@@ -226,14 +250,19 @@ function fullImport(
   let duplicateCount = 0
   const dedupedMessages: PushImportMessage[] = []
   for (const msg of messages) {
+    const key = generateMessageKey(msg.timestamp, msg.sender, msg.content ?? null)
     if (msg.platformMessageId) {
       if (pmids.has(msg.platformMessageId)) {
         duplicateCount++
         continue
       }
+      if (hashKeys.has(key)) {
+        duplicateCount++
+        continue
+      }
       pmids.add(msg.platformMessageId)
+      hashKeys.add(key)
     } else {
-      const key = generateMessageKey(msg.timestamp, msg.sender, msg.content ?? null)
       if (hashKeys.has(key)) {
         duplicateCount++
         continue
@@ -248,11 +277,15 @@ function fullImport(
   const stats = writeParseResultToDb(
     db,
     meta,
-    allMembers.map((m) => ({ ...m, accountName: m.accountName ?? m.platformId })),
+    allMembers.map((m) => ({
+      ...m,
+      accountName: normalizeSenderAccountName(m.platformId, m.accountName) ?? m.platformId,
+      groupNickname: normalizeSenderGroupNickname(m.platformId, m.groupNickname),
+    })),
     dedupedMessages.map((m) => ({
       senderPlatformId: m.sender,
-      senderAccountName: m.accountName ?? m.sender,
-      senderGroupNickname: m.groupNickname,
+      senderAccountName: normalizeSenderAccountName(m.sender, m.accountName) ?? m.sender,
+      senderGroupNickname: normalizeSenderGroupNickname(m.sender, m.groupNickname),
       timestamp: m.timestamp,
       type: m.type,
       content: m.content ?? null,
@@ -331,10 +364,12 @@ function incrementalImport(
     db.transaction(() => {
       for (const m of payload.members!) {
         const existed = existingMemberIds.has(m.platformId)
+        const accountName = normalizeSenderAccountName(m.platformId, m.accountName)
+        const groupNickname = normalizeSenderGroupNickname(m.platformId, m.groupNickname)
         upsertMember.run(
           m.platformId,
-          m.accountName || null,
-          m.groupNickname || null,
+          accountName || null,
+          groupNickname || null,
           m.avatar || null,
           m.roles ? JSON.stringify(m.roles) : '[]'
         )

packages/sync/src/pull-engine.test.ts

@@ -69,6 +69,7 @@ function createEngine(options: {
   files: string[]
   importResult: Awaited<ReturnType<DataImporter['importFile']>>
   dataSource: DataSource
+  isImporting?: (sessionId: string | undefined) => boolean
   fetchParams?: FetchParams[]
   sessionUpdates?: Array<{ sessionId: string; updates: Partial<ImportSession> }>
   pullResults?: Array<{ status: 'success' | 'error'; detail: string }>
@@ -109,6 +110,7 @@ function createEngine(options: {
     importer,
     notifier,
     dsManager: dsManager as any,
+    isImporting: options.isImporting,
   })
 }
 
@@ -146,6 +148,40 @@ describe('PullEngine', () => {
     assert.equal(importCount, 1)
   })
 
+  it('does not skip pulls when a different local session is importing', async () => {
+    const session = createSession()
+    session.targetSessionId = 'local_session_b'
+    const dataSource = createDataSource()
+    dataSource.sessions = [session]
+    const page = writeTempJson({
+      chatlab: { version: '0.0.2', exportedAt: 100 },
+      meta: { name: 'Test Session', platform: 'test', type: 'group' },
+      members: [{ platformId: 'u1', accountName: 'Alice' }],
+      messages: [{ sender: 'u1', timestamp: 101, type: 0, content: 'hi' }],
+      sync: { hasMore: false, nextSince: 101 },
+    })
+    const checkedSessionIds: Array<string | undefined> = []
+    const engine = createEngine({
+      files: [page],
+      dataSource,
+      isImporting: (sessionId) => {
+        checkedSessionIds.push(sessionId)
+        return sessionId === 'local_session_a'
+      },
+      importResult: {
+        success: true,
+        newMessageCount: 1,
+        sessionId: session.targetSessionId,
+      },
+    })
+
+    const result = await withImmediateTimers(() => engine.executePullSession(dataSource.id, dataSource, session))
+
+    assert.equal(result.success, true)
+    assert.equal(result.newMessageCount, 1)
+    assert.deepEqual(checkedSessionIds, ['local_session_b'])
+  })
+
   it('reports retry import failure instead of marking the pull successful', async () => {
     const session = createSession()
     const dataSource = createDataSource()

packages/sync/src/pull-engine.ts

@@ -156,8 +156,8 @@ export interface PullEngineOptions {
   notifier: SyncNotifier
   dsManager: DataSourceManager
   logger?: SyncLogger
-  /** Return true when an import is already in progress (skip pull) */
-  isImporting?: () => boolean
+  /** Return true when an import is already in progress for the local session (skip pull) */
+  isImporting?: (sessionId: string | undefined) => boolean
   /**
    * Called after a pull session completes successfully with the local session ID.
    * Used for post-import side effects (e.g. applying the platform owner profile).
@@ -172,7 +172,7 @@ export class PullEngine {
   private notifier: SyncNotifier
   private dsManager: DataSourceManager
   private logger: SyncLogger
-  private isImporting: () => boolean
+  private isImporting: (sessionId: string | undefined) => boolean
   private onSessionImported?: (localSessionId: string) => void
   private pullingSourceIds = new Set<string>()
   private progressMap = new Map<string, PullProgress>()
@@ -222,7 +222,8 @@ export class PullEngine {
       return { success: true, newMessageCount: 0 }
     }
 
-    if (this.isImporting()) {
+    const importStatusSessionId = sess.targetSessionId || deriveLocalSessionId(ds.baseUrl, sess.remoteSessionId)
+    if (this.isImporting(importStatusSessionId)) {
       this.logger.info(`[Pull] Skipping "${sess.name}": import in progress`)
       return { success: false, newMessageCount: 0, error: 'Import in progress' }
     }