fix(import): harden automatic incremental imports

apps/cli/src/cli.ts

@@ -13,6 +13,7 @@ import {
   initAppLogger,
   appLogger,
   getSystemLogsDir,
+  IMPORT_IN_PROGRESS_ERROR_KEY,
   logNativeParserStatus,
 } from '@openchatlab/node-runtime'
 import { getVersion } from './version'
@@ -95,8 +96,12 @@ program
       if (result.success) {
         console.log(`\n\nImport succeeded!`)
         console.log(`  Session ID: ${result.sessionId}`)
-        console.log(`  Mode: ${result.importMode === 'incremental' ? 'incremental' : 'created'}`)
+        console.log(`  Mode: ${result.importMode ?? 'unknown'}`)
         if (result.matchedBy) console.log(`  Matched by: ${result.matchedBy}`)
+        if (result.createReason) console.log(`  Create reason: ${result.createReason}`)
+        if (result.createReason === 'ambiguous') {
+          console.warn('  Warning: multiple existing sessions matched, so a new session was created.')
+        }
         console.log(`  New messages: ${result.newMessageCount ?? 0}`)
         console.log(`  Duplicates skipped: ${result.duplicateCount ?? 0}`)
 
@@ -117,7 +122,11 @@ program
           }
         }
       } else {
-        console.error(`\n\nImport failed: ${result.error}`)
+        const errorMessage =
+          result.error === IMPORT_IN_PROGRESS_ERROR_KEY
+            ? 'Another import is already in progress. Please retry later.'
+            : result.error
+        console.error(`\n\nImport failed: ${errorMessage}`)
         process.exitCode = 1
       }
     } catch (err) {

apps/cli/src/http/routes/web/import.test.ts

@@ -71,4 +71,69 @@ describe('CLI Web automatic import route', () => {
       await app.close()
     }
   })
+
+  it('returns a stable import-in-progress error event instead of an internal database error', async () => {
+    const app = Fastify()
+    try {
+      await app.register(multipart)
+      registerImportRoutes(
+        app,
+        {} as any,
+        {
+          runAutoImport: async () => ({ success: false, error: 'error.import_in_progress' }),
+        } as any
+      )
+
+      const body = multipartPayload()
+      const response = await app.inject({
+        method: 'POST',
+        url: '/_web/import',
+        headers: { 'content-type': body.contentType },
+        payload: body.payload,
+      })
+
+      assert.equal(response.statusCode, 200)
+      assert.match(response.body, /event: error/)
+      assert.match(response.body, /"error":"error\.import_in_progress"/)
+      assert.doesNotMatch(response.body, /no such column|SQLITE/i)
+    } finally {
+      await app.close()
+    }
+  })
+
+  it('preserves an ambiguous create reason in the done event', async () => {
+    const app = Fastify()
+    try {
+      await app.register(multipart)
+      registerImportRoutes(
+        app,
+        {} as any,
+        {
+          runAutoImport: async () => ({
+            success: true,
+            sessionId: 'new-session',
+            importMode: 'created',
+            createReason: 'ambiguous',
+            newMessageCount: 3,
+            duplicateCount: 0,
+          }),
+        } as any
+      )
+
+      const body = multipartPayload()
+      const response = await app.inject({
+        method: 'POST',
+        url: '/_web/import',
+        headers: { 'content-type': body.contentType },
+        payload: body.payload,
+      })
+
+      assert.equal(response.statusCode, 200)
+      assert.match(response.body, /event: done/)
+      assert.match(response.body, /"importMode":"created"/)
+      assert.match(response.body, /"createReason":"ambiguous"/)
+    } finally {
+      await app.close()
+    }
+  })
 })

apps/cli/src/http/routes/web/index.ts

@@ -98,6 +98,11 @@ export function registerWebRoutes(
   const cliStreamImport = async (dm: typeof dbManager, filePath: string) => {
     const { streamImport } = await import('../../../import/stream-import')
     const result = await streamImport(dm, filePath)
+    if (!result.success) {
+      const { IMPORT_IN_PROGRESS_ERROR_KEY, ImportInProgressError } = await import('@openchatlab/node-runtime')
+      if (result.error === IMPORT_IN_PROGRESS_ERROR_KEY) throw new ImportInProgressError()
+      throw new Error(result.error || 'Import failed')
+    }
     if (!result.sessionId) throw new Error('Import succeeded but no sessionId returned')
     return { sessionId: result.sessionId }
   }

apps/cli/src/import/importer.test.ts

@@ -4,7 +4,13 @@ import os from 'node:os'
 import path from 'node:path'
 import test from 'node:test'
 import type { PathProvider } from '@openchatlab/core'
-import { DatabaseManager, raiseDataDirMinRuntimeVersion, readDataDirCompatibilityMeta } from '@openchatlab/node-runtime'
+import {
+  DatabaseManager,
+  IMPORT_IN_PROGRESS_ERROR_KEY,
+  raiseDataDirMinRuntimeVersion,
+  readDataDirCompatibilityMeta,
+  withDataDirImportLock,
+} from '@openchatlab/node-runtime'
 import { autoImport, streamImport } from './stream-import'
 
 const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node')
@@ -45,6 +51,21 @@ function writeTempChatFile(dir: string): string {
   return filePath
 }
 
+function writeTempDuplicateChatFile(dir: string): string {
+  const filePath = path.join(dir, 'duplicate-chat.json')
+  const message = { sender: 'u1', accountName: 'Alice', timestamp: 1711468800, type: 0, content: 'hello' }
+  fs.writeFileSync(
+    filePath,
+    JSON.stringify({
+      chatlab: { version: '0.0.2', exportedAt: 1711468800 },
+      meta: { name: 'Duplicate Chat', platform: 'qq', type: 'group', groupId: 'duplicate-group' },
+      members: [{ platformId: 'u1', accountName: 'Alice' }],
+      messages: [message, { ...message }],
+    })
+  )
+  return filePath
+}
+
 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 })
@@ -115,6 +136,32 @@ test('autoImport creates once and then incrementally imports the same stable cha
   assert.equal(manager.listSessionIds().length, 1)
 })
 
+test('autoImport reports the same exact-message deduplication on create and incremental paths', async (t) => {
+  const root = makeTempDir()
+  t.after(() => fs.rmSync(root, { recursive: true, force: true }))
+  fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true })
+  const manager = new DatabaseManager(createPathProvider(root), {
+    nativeBinding,
+    runtime: { version: '0.25.1', kind: 'cli' },
+  })
+  const chatFile = writeTempDuplicateChatFile(root)
+
+  const created = await autoImport(manager, chatFile, { nativeBinding })
+  const incremental = await autoImport(manager, chatFile, { nativeBinding })
+
+  assert.equal(created.success, true)
+  assert.equal(created.newMessageCount, 1)
+  assert.equal(created.duplicateCount, 1)
+  assert.equal(incremental.success, true)
+  assert.equal(incremental.newMessageCount, 0)
+  assert.equal(incremental.duplicateCount, 2)
+
+  const db = manager.openRawSessionDatabase(created.sessionId!, { readonly: true })
+  const row = db.prepare('SELECT COUNT(*) AS count FROM message').get() as { count: number }
+  db.close()
+  assert.equal(row.count, 1)
+})
+
 test('autoImport uses an explicit id for create and then forces incremental import', async () => {
   const root = makeTempDir()
   fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true })
@@ -136,3 +183,19 @@ test('autoImport uses an explicit id for create and then forces incremental impo
   assert.equal(incremental.newMessageCount, 0)
   assert.equal(manager.listSessionIds().length, 1)
 })
+
+test('autoImport rejects a concurrent writer before opening a session database', async (t) => {
+  const root = makeTempDir()
+  t.after(() => fs.rmSync(root, { recursive: true, force: true }))
+  fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true })
+  const manager = new DatabaseManager(createPathProvider(root), {
+    nativeBinding,
+    runtime: { version: '0.25.1', kind: 'cli' },
+  })
+  const chatFile = writeTempChatFile(root)
+
+  const result = await withDataDirImportLock(manager.getUserDataDir(), () => autoImport(manager, chatFile))
+
+  assert.deepEqual(result, { success: false, error: IMPORT_IN_PROGRESS_ERROR_KEY })
+  assert.equal(fs.readdirSync(path.join(root, 'data', 'databases')).length, 0)
+})

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

@@ -9,11 +9,14 @@
 import type { DatabaseManager } from '@openchatlab/node-runtime'
 import {
   DataDirCompatibilityError,
+  IMPORT_IN_PROGRESS_ERROR_KEY,
+  ImportInProgressError,
   autoImportFile as sharedAutoImportFile,
   streamingImport,
   analyzeNewImport as sharedAnalyzeNewImport,
   analyzeIncrementalImport as sharedAnalyzeIncremental,
   incrementalImport as sharedIncrementalImport,
+  withDataDirImportLock,
 } from '@openchatlab/node-runtime'
 import type {
   StreamImportResult,
@@ -128,7 +131,7 @@ function deleteSessionDatabase(dbManager: DatabaseManager, sessionId: string): v
  * High-performance streaming import: parse a file and write to DB
  * with batched transactions, deferred indexes, and FTS.
  */
-export async function streamImport(
+async function streamImportUnlocked(
   dbManager: DatabaseManager,
   filePath: string,
   options?: StreamImportOptions
@@ -159,6 +162,23 @@ export async function streamImport(
   return result
 }
 
+export async function streamImport(
+  dbManager: DatabaseManager,
+  filePath: string,
+  options?: StreamImportOptions
+): Promise<StreamImportResult> {
+  try {
+    return await withDataDirImportLock(dbManager.getUserDataDir(), () =>
+      streamImportUnlocked(dbManager, filePath, options)
+    )
+  } catch (error) {
+    if (error instanceof ImportInProgressError) {
+      return { success: false, error: IMPORT_IN_PROGRESS_ERROR_KEY }
+    }
+    throw error
+  }
+}
+
 export async function autoImport(
   dbManager: DatabaseManager,
   filePath: string,
@@ -170,30 +190,40 @@ export async function autoImport(
   if (chatIndex !== undefined) formatOptions.chatIndex = chatIndex
   const progressAdapter = createProgressAdapter(onProgress)
 
-  return sharedAutoImportFile(
-    filePath,
-    {
-      listSessionIds: () => dbManager.listSessionIds(),
-      openReadonly: (candidateSessionId) => dbManager.openRawSessionDatabase(candidateSessionId, { readonly: true }),
-      onProgress: progressAdapter,
-      sessionExists: (candidateSessionId) => dbManager.listSessionIds().includes(candidateSessionId),
-      createSession: (sourcePath, sourceFormatOptions, explicitSessionId) =>
-        streamImport(dbManager, sourcePath, {
-          ...options,
-          ...sourceFormatOptions,
-          sessionId: explicitSessionId,
-        }),
-      appendSession: (targetSessionId, sourcePath, sourceFormatOptions) =>
-        incrementalImport(dbManager, targetSessionId, sourcePath, {
-          ...sourceFormatOptions,
+  try {
+    return await withDataDirImportLock(dbManager.getUserDataDir(), () =>
+      sharedAutoImportFile(
+        filePath,
+        {
+          listSessionIds: () => dbManager.listSessionIds(),
+          openReadonly: (candidateSessionId) =>
+            dbManager.openRawSessionDatabase(candidateSessionId, { readonly: true }),
           onProgress: progressAdapter,
-        }),
-    },
-    {
-      explicitSessionId: sessionId,
-      formatOptions,
+          sessionExists: (candidateSessionId) => dbManager.listSessionIds().includes(candidateSessionId),
+          createSession: (sourcePath, sourceFormatOptions, explicitSessionId) =>
+            streamImportUnlocked(dbManager, sourcePath, {
+              ...options,
+              ...sourceFormatOptions,
+              sessionId: explicitSessionId,
+            }),
+          appendSession: (targetSessionId, sourcePath, sourceFormatOptions) =>
+            incrementalImportUnlocked(dbManager, targetSessionId, sourcePath, {
+              ...sourceFormatOptions,
+              onProgress: progressAdapter,
+            }),
+        },
+        {
+          explicitSessionId: sessionId,
+          formatOptions,
+        }
+      )
+    )
+  } catch (error) {
+    if (error instanceof ImportInProgressError) {
+      return { success: false, error: IMPORT_IN_PROGRESS_ERROR_KEY }
     }
-  )
+    throw error
+  }
 }
 
 // ==================== Incremental import ====================
@@ -216,7 +246,7 @@ function buildIncrementalDeps(
   }
 }
 
-export async function incrementalImport(
+async function incrementalImportUnlocked(
   dbManager: DatabaseManager,
   sessionId: string,
   filePath: string,
@@ -249,6 +279,24 @@ export async function incrementalImport(
   return result
 }
 
+export async function incrementalImport(
+  dbManager: DatabaseManager,
+  sessionId: string,
+  filePath: string,
+  options?: ImportOptions & { onProgress?: ImportProgressCallback }
+): Promise<IncrementalImportResult> {
+  try {
+    return await withDataDirImportLock(dbManager.getUserDataDir(), () =>
+      incrementalImportUnlocked(dbManager, sessionId, filePath, options)
+    )
+  } catch (error) {
+    if (error instanceof ImportInProgressError) {
+      return { success: false, newMessageCount: 0, error: IMPORT_IN_PROGRESS_ERROR_KEY }
+    }
+    throw error
+  }
+}
+
 export async function analyzeIncrementalImport(
   dbManager: DatabaseManager,
   sessionId: string,

apps/desktop/main/api/errors.test.ts

@@ -2,6 +2,14 @@ import assert from 'node:assert/strict'
 import test from 'node:test'
 import { DataDirCompatibilityError } from '@openchatlab/node-runtime/src/data-dir-compat'
 import { ApiErrorCode, apiErrorFromUnknown } from './errors'
+import { ImportInProgressError } from '@openchatlab/node-runtime/src/import/import-lock'
+
+test('apiErrorFromUnknown maps the shared import lock error to 409', () => {
+  const apiError = apiErrorFromUnknown(new ImportInProgressError())
+
+  assert.equal(apiError?.code, ApiErrorCode.IMPORT_IN_PROGRESS)
+  assert.equal(apiError?.statusCode, 409)
+})
 
 test('apiErrorFromUnknown maps wrapped data directory compatibility errors to 409', () => {
   const cause = new DataDirCompatibilityError(

apps/desktop/main/api/errors.ts

@@ -3,6 +3,7 @@
  */
 
 import { DataDirCompatibilityError } from '@openchatlab/node-runtime/src/data-dir-compat'
+import { ImportInProgressError } from '@openchatlab/node-runtime/src/import/import-lock'
 
 export enum ApiErrorCode {
   UNAUTHORIZED = 'UNAUTHORIZED',
@@ -103,6 +104,7 @@ export function serverError(message = 'Internal server error'): ApiError {
 
 export function apiErrorFromUnknown(error: unknown): ApiError | null {
   if (error instanceof ApiError) return error
+  if (error instanceof ImportInProgressError) return importInProgress()
   const compatibilityError = findDataDirCompatibilityError(error)
   if (compatibilityError) return dataDirIncompatible(compatibilityError.message)
   return null

apps/desktop/main/internal-api.ts

@@ -21,6 +21,7 @@ import {
   CustomProviderStore,
   CustomModelStore,
   MergeSessionCache,
+  withDataDirImportLock,
   raiseChatDbCompatibilityGate,
   streamingImport,
   createSemanticIndexWorkerRuntimeClient,
@@ -138,26 +139,29 @@ export async function startInternalServer(pathProvider: PathProvider): Promise<I
     }
 
     const electronStreamImport = async (dm: DatabaseManager, filePath: string) => {
-      const deps: StreamImportDeps = {
-        openDatabase(sessionId: string) {
-          return dm.openRawSessionDatabase(sessionId, { create: true, initializeChatTables: true })
-        },
-        deleteDatabase(sessionId: string) {
-          dm.deleteSessionDatabaseFiles(sessionId)
-        },
-        onProgress() {
-          /* no progress for merge-triggered import */
-        },
-      }
-      const result = await streamingImport(filePath, deps)
-      if (!result.sessionId) throw new Error('Import succeeded but no sessionId returned')
-      try {
-        raiseChatDbCompatibilityGate(pathProvider, runtime)
-      } catch (error) {
-        dm.deleteSessionDatabaseFiles(result.sessionId)
-        throw error
-      }
-      return { sessionId: result.sessionId }
+      return withDataDirImportLock(dm.getUserDataDir(), async () => {
+        const deps: StreamImportDeps = {
+          openDatabase(sessionId: string) {
+            return dm.openRawSessionDatabase(sessionId, { create: true, initializeChatTables: true })
+          },
+          deleteDatabase(sessionId: string) {
+            dm.deleteSessionDatabaseFiles(sessionId)
+          },
+          onProgress() {
+            /* no progress for merge-triggered import */
+          },
+        }
+        const result = await streamingImport(filePath, deps)
+        if (!result.success) throw new Error(result.error || 'Import failed')
+        if (!result.sessionId) throw new Error('Import succeeded but no sessionId returned')
+        try {
+          raiseChatDbCompatibilityGate(pathProvider, runtime)
+        } catch (error) {
+          dm.deleteSessionDatabaseFiles(result.sessionId)
+          throw error
+        }
+        return { sessionId: result.sessionId }
+      })
     }
 
     const { shell } = await import('electron')

apps/desktop/main/worker/workerManager.ts

@@ -15,7 +15,12 @@ import { getNlpDir } from '../nlp/dictManager'
 import { resolveDesktopNativeBinding } from '../native-sqlite'
 import { assertDesktopDataDirCompatible, getDesktopAppVersion } from '../runtime-compat'
 import { getPathProvider } from '../path-context'
-import { raiseChatDbCompatibilityGate } from '@openchatlab/node-runtime'
+import {
+  IMPORT_IN_PROGRESS_ERROR_KEY,
+  ImportInProgressError,
+  raiseChatDbCompatibilityGate,
+  withDataDirImportLock,
+} from '@openchatlab/node-runtime'
 import { isRestartableReadOnlyRequestType } from './workerTimeoutPolicy'
 
 interface WorkerRequestOptions {
@@ -519,7 +524,7 @@ export async function streamParseFileInfo(
  * @param onProgress 进度回调
  * @param formatOptions 格式特定选项（如 Telegram 的 chatIndex）
  */
-export async function streamImport(
+async function streamImportUnlocked(
   filePath: string,
   onProgress?: (progress: ParseProgress) => void,
   formatOptions?: Record<string, unknown>,
@@ -551,36 +556,63 @@ export async function streamImport(
   return result
 }
 
+export async function streamImport(
+  filePath: string,
+  onProgress?: (progress: ParseProgress) => void,
+  formatOptions?: Record<string, unknown>,
+  externalSessionId?: string
+): Promise<StreamImportResult> {
+  try {
+    return await withDataDirImportLock(getPathProvider().getUserDataDir(), () =>
+      streamImportUnlocked(filePath, onProgress, formatOptions, externalSessionId)
+    )
+  } catch (error) {
+    if (error instanceof ImportInProgressError) {
+      return { success: false, error: IMPORT_IN_PROGRESS_ERROR_KEY }
+    }
+    throw error
+  }
+}
+
 export async function autoImport(
   filePath: string,
   onProgress?: (progress: ParseProgress) => void,
   formatOptions?: Record<string, unknown>,
   explicitSessionId?: string
 ): Promise<AutoImportResult> {
-  assertDataDirCompatibleNow()
-
-  const result = await sendToWorkerWithProgress<AutoImportResult>(
-    'autoImport',
-    { filePath, formatOptions, explicitSessionId },
-    onProgress
-  )
-  if (!result.success || !result.sessionId || result.importMode !== 'created') return result
-
   try {
-    raiseChatDbCompatibilityGate(getPathProvider(), {
-      version: getDesktopAppVersion(app.getVersion()),
-      kind: 'desktop',
+    return await withDataDirImportLock(getPathProvider().getUserDataDir(), async () => {
+      assertDataDirCompatibleNow()
+
+      const result = await sendToWorkerWithProgress<AutoImportResult>(
+        'autoImport',
+        { filePath, formatOptions, explicitSessionId },
+        onProgress
+      )
+      if (!result.success || !result.sessionId || result.importMode !== 'created') return result
+
+      try {
+        raiseChatDbCompatibilityGate(getPathProvider(), {
+          version: getDesktopAppVersion(app.getVersion()),
+          kind: 'desktop',
+        })
+      } catch (error) {
+        deleteImportedSessionFiles(result.sessionId)
+        return {
+          success: false,
+          error: error instanceof Error ? error.message : String(error),
+          diagnostics: result.diagnostics,
+        }
+      }
+
+      return result
     })
   } catch (error) {
-    deleteImportedSessionFiles(result.sessionId)
-    return {
-      success: false,
-      error: error instanceof Error ? error.message : String(error),
-      diagnostics: result.diagnostics,
+    if (error instanceof ImportInProgressError) {
+      return { success: false, error: IMPORT_IN_PROGRESS_ERROR_KEY }
     }
+    throw error
   }
-
-  return result
 }
 
 function deleteImportedSessionFiles(sessionId: string): void {
@@ -981,9 +1013,17 @@ export async function incrementalImport(
   onProgress?: (progress: ParseProgress) => void,
   options?: ImportOptions
 ): Promise<IncrementalImportResult> {
-  assertDataDirCompatibleNow()
-
-  return sendToWorkerWithProgress('incrementalImport', { sessionId, filePath, options }, onProgress)
+  try {
+    return await withDataDirImportLock(getPathProvider().getUserDataDir(), async () => {
+      assertDataDirCompatibleNow()
+      return sendToWorkerWithProgress('incrementalImport', { sessionId, filePath, options }, onProgress)
+    })
+  } catch (error) {
+    if (error instanceof ImportInProgressError) {
+      return { success: false, newMessageCount: 0, error: IMPORT_IN_PROGRESS_ERROR_KEY }
+    }
+    throw error
+  }
 }
 
 /**

apps/desktop/preload/apis/chat.ts

@@ -13,6 +13,7 @@ interface ChatImportResult {
   error?: string
   importMode?: 'created' | 'incremental'
   matchedBy?: 'source-session-id' | 'stable-id' | 'trailing-messages'
+  createReason?: 'no-match' | 'ambiguous'
   newMessageCount?: number
   duplicateCount?: number
   diagnostics?: {
@@ -20,6 +21,7 @@ interface ChatImportResult {
     detectedFormat: string | null
     messagesReceived: number
     messagesWritten: number
+    duplicateCount: number
     messagesSkipped: number
     skipReasons: { noSenderId: number; noAccountName: number; invalidTimestamp: number; noType: number }
   }

apps/desktop/preload/index.d.ts

@@ -27,6 +27,8 @@ interface ImportDiagnostics {
   messagesReceived: number
   /** 写入的消息数 */
   messagesWritten: number
+  /** 去重过滤的消息数 */
+  duplicateCount: number
   /** 跳过的消息数 */
   messagesSkipped: number
   /** 跳过原因统计 */
@@ -44,6 +46,7 @@ interface ChatImportResult {
   error?: string
   importMode?: 'created' | 'incremental'
   matchedBy?: 'source-session-id' | 'stable-id' | 'trailing-messages'
+  createReason?: 'no-match' | 'ambiguous'
   newMessageCount?: number
   duplicateCount?: number
   diagnostics?: ImportDiagnostics