fix(import): align shared push import semantics

apps/desktop/main/worker/dbWorker.ts

@@ -81,6 +81,7 @@ import {
   analyzeIncrementalImport,
   incrementalImport,
   analyzeNewImport,
+  pushImport,
 } from './import'
 import { initNlpDir } from '@openchatlab/node-runtime'
 
@@ -248,6 +249,7 @@ const asyncHandlers: Record<string, (payload: any, requestId: string) => Promise
   // 流式导入
   streamImport: (p, id) => streamImport(p.filePath, id, p.formatOptions, p.externalSessionId),
   autoImport: (p, id) => autoImport(p.filePath, id, p.formatOptions, p.explicitSessionId),
+  pushImport: (p) => pushImport(p.sessionId, p.payload),
   // 流式解析文件信息（用于合并预览）
   streamParseFileInfo: (p, id) => streamParseFileInfo(p.filePath, id),
   // 增量导入

apps/desktop/main/worker/import/index.ts

@@ -24,6 +24,8 @@ export {
   type IncrementalImportResult,
 } from './incrementalImport'
 
+export { pushImport } from './pushImport'
+
 // 工具函数（供其他模块使用）
 export { sendProgress, generateSessionId, getDbPath, createDatabaseWithoutIndexes, createIndexes } from './utils'
 

apps/desktop/main/worker/import/pushImport.test.ts

@@ -0,0 +1,37 @@
+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 { initDbDir } from '../core/dbCore'
+import { pushImport } from './pushImport'
+
+const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node')
+
+test('Desktop worker push import uses the shared writer and canonical database directory', async (t) => {
+  const baseDir = fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir()
+  const root = fs.mkdtempSync(path.join(baseDir, 'chatlab-worker-push-import-'))
+  t.after(() => fs.rmSync(root, { recursive: true, force: true }))
+
+  const dbDir = path.join(root, 'databases')
+  initDbDir(dbDir, path.join(root, 'cache'), path.join(root, 'temp'), nativeBinding, path.join(root, 'logs'))
+
+  const outcome = await pushImport('worker-session', {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Worker Push Import', platform: 'wechat', type: 'private' },
+    members: [{ platformId: 'wxid_alice', accountName: 'Alice' }],
+    messages: [
+      {
+        platformMessageId: 'msg-1',
+        sender: 'wxid_alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'hello',
+      },
+    ],
+  })
+
+  assert.equal(outcome.ok, true, JSON.stringify(outcome))
+  assert.equal(fs.existsSync(path.join(dbDir, 'worker-session.db')), true)
+  assert.equal(fs.existsSync(path.join(root, 'worker-session.db')), false)
+})

apps/desktop/main/worker/import/pushImport.ts

@@ -0,0 +1,43 @@
+import * as fs from 'node:fs'
+import * as path from 'node:path'
+import { BetterSqliteAdapter } from '@openchatlab/node-runtime/src/better-sqlite3-adapter'
+import {
+  executePushImportUnlocked,
+  type PushImportOutcome,
+  type PushImportPayload,
+} from '@openchatlab/node-runtime/src/services/push-importer'
+import { closeDatabase, getDbDir, openRawDatabase } from '../core/dbCore'
+
+function getDbPath(sessionId: string): string {
+  return path.join(getDbDir(), `${sessionId}.db`)
+}
+
+export async function pushImport(sessionId: string, payload: PushImportPayload): Promise<PushImportOutcome> {
+  return executePushImportUnlocked(
+    {
+      getDbPath,
+      openDatabase(id, options) {
+        const dbPath = getDbPath(id)
+        if (!options.create && !fs.existsSync(dbPath)) throw new Error(`Session database not found: ${id}`)
+        closeDatabase(id)
+        fs.mkdirSync(path.dirname(dbPath), { recursive: true })
+        const db = options.readonly === undefined ? openRawDatabase(dbPath) : openRawDatabase(dbPath, options)
+        return new BetterSqliteAdapter(db)
+      },
+      deleteDatabase(id) {
+        closeDatabase(id)
+        const dbPath = getDbPath(id)
+        for (const suffix of ['', '-wal', '-shm']) {
+          try {
+            const filePath = dbPath + suffix
+            if (fs.existsSync(filePath)) fs.unlinkSync(filePath)
+          } catch {
+            /* best-effort cleanup */
+          }
+        }
+      },
+    },
+    sessionId,
+    payload
+  )
+}

apps/desktop/main/worker/workerManager.ts

@@ -21,6 +21,7 @@ import {
   raiseChatDbCompatibilityGate,
   withDataDirImportLock,
 } from '@openchatlab/node-runtime'
+import type { PushImportOutcome, PushImportPayload } from '@openchatlab/node-runtime'
 import { isRestartableReadOnlyRequestType } from './workerTimeoutPolicy'
 
 interface WorkerRequestOptions {
@@ -615,6 +616,40 @@ export async function autoImport(
   }
 }
 
+/**
+ * 在 worker 中执行共享 Push Import 写库语义；主进程持有全局导入锁，
+ * 并在 worker 成功后完成数据目录兼容门禁更新。
+ */
+export async function pushImport(sessionId: string, payload: PushImportPayload): Promise<PushImportOutcome> {
+  try {
+    return await withDataDirImportLock(getPathProvider().getUserDataDir(), async () => {
+      assertDataDirCompatibleNow()
+      const outcome = await sendToWorker<PushImportOutcome>('pushImport', { sessionId, payload })
+      if (!outcome.ok) return outcome
+
+      try {
+        raiseChatDbCompatibilityGate(getPathProvider(), {
+          version: getDesktopAppVersion(app.getVersion()),
+          kind: 'desktop',
+        })
+      } catch (error) {
+        if (outcome.result.created) deleteImportedSessionFiles(sessionId)
+        throw error
+      }
+      return outcome
+    })
+  } catch (error) {
+    if (error instanceof ImportInProgressError) {
+      return {
+        ok: false,
+        reason: 'import_in_progress',
+        message: 'Another import is already in progress',
+      }
+    }
+    throw error
+  }
+}
+
 function deleteImportedSessionFiles(sessionId: string): void {
   const dbPath = path.join(getDbDir(), `${sessionId}.db`)
   for (const suffix of ['', '-wal', '-shm']) {

apps/desktop/main/worker/workerTimeoutPolicy.test.ts

@@ -21,6 +21,7 @@ describe('worker timeout policy', () => {
     assert.equal(isRestartableReadOnlyRequestType('deleteMember'), false)
     assert.equal(isRestartableReadOnlyRequestType('streamImport'), false)
     assert.equal(isRestartableReadOnlyRequestType('autoImport'), false)
+    assert.equal(isRestartableReadOnlyRequestType('pushImport'), false)
     assert.equal(isRestartableReadOnlyRequestType('incrementalImport'), false)
     assert.equal(isRestartableReadOnlyRequestType('exportFilterResultToFile'), false)
     assert.equal(isRestartableReadOnlyRequestType('unknownFutureWorkerRequest'), false)

packages/node-runtime/src/import/message-deduplicator.ts

@@ -35,8 +35,8 @@ export function createMessageDedupState(
 }
 
 /**
- * Canonical file-import dedup rule: prefer stable platform message IDs and
- * fall back to timestamp + sender platform ID + type + normalized content + reply target.
+ * Canonical import dedup rule shared by file imports and JSON Push: prefer stable platform
+ * message IDs and fall back to timestamp + sender platform ID + type + normalized content + reply target.
  */
 export function registerMessageAndCheckDuplicate(message: DedupMessage, state: MessageDedupState): boolean {
   const key = generateFallbackMessageKey(message)

packages/node-runtime/src/import/session-id.ts

@@ -0,0 +1,5 @@
+const SAFE_IMPORT_SESSION_ID_RE = /^[A-Za-z0-9_@-][A-Za-z0-9._@-]{0,127}$/
+
+export function isValidImportSessionId(sessionId: string): boolean {
+  return SAFE_IMPORT_SESSION_ID_RE.test(sessionId) && !sessionId.includes('..')
+}

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

@@ -4,6 +4,7 @@ import os from 'node:os'
 import path from 'node:path'
 import test from 'node:test'
 import { DatabaseManager } from '../database-manager'
+import { raiseDataDirMinRuntimeVersion, readDataDirCompatibilityMeta, type RuntimeIdentity } from '../data-dir-compat'
 import { withDataDirImportLock } from '../import/import-lock'
 import type { PushImportPayload } from './push-importer'
 import { pushImport } from './push-importer'
@@ -15,7 +16,7 @@ function makeTempDir(): string {
   return fs.mkdtempSync(path.join(baseDir, 'chatlab-push-import-'))
 }
 
-function createDatabaseManager(rootDir: string): DatabaseManager {
+function createDatabaseManager(rootDir: string, runtime?: RuntimeIdentity): DatabaseManager {
   return new DatabaseManager(
     {
       getSystemDir: () => rootDir,
@@ -29,7 +30,7 @@ function createDatabaseManager(rootDir: string): DatabaseManager {
       getLogsDir: () => path.join(rootDir, 'logs'),
       getDownloadsDir: () => rootDir,
     },
-    { nativeBinding, allowMissingRuntimeForTests: true }
+    runtime ? { nativeBinding, runtime } : { nativeBinding, allowMissingRuntimeForTests: true }
   )
 }
 
@@ -54,6 +55,81 @@ test('rejects push imports while any writer holds the data-directory import lock
   assert.equal(fs.existsSync(manager.getDbPath('different-session')), false)
 })
 
+test('rejects unsafe session IDs before resolving or opening a database path', async (t) => {
+  const root = makeTempDir()
+  t.after(() => fs.rmSync(root, { recursive: true, force: true }))
+  const manager = createDatabaseManager(root)
+
+  const outcome = await pushImport(manager, '../escape', {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Unsafe Session', platform: 'wechat', type: 'private' },
+    messages: [{ sender: 'wxid_alice', timestamp: 1780330832, type: 0, content: 'hello' }],
+  })
+
+  assert.deepEqual(outcome, {
+    ok: false,
+    reason: 'invalid_payload',
+    message: 'sessionId contains invalid characters',
+  })
+  assert.equal(fs.existsSync(path.join(root, 'escape.db')), false)
+})
+
+test('raises the data directory compatibility gate after a successful push import', async (t) => {
+  const root = makeTempDir()
+  t.after(() => fs.rmSync(root, { recursive: true, force: true }))
+  const manager = createDatabaseManager(root, { version: '0.25.1', kind: 'cli' })
+
+  const outcome = await pushImport(manager, 'compatibility-gate', {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Compatibility Gate', platform: 'wechat', type: 'private' },
+    messages: [{ sender: 'wxid_alice', timestamp: 1780330832, type: 0, content: 'hello' }],
+  })
+
+  assert.equal(outcome.ok, true)
+  const meta = readDataDirCompatibilityMeta(root)
+  assert.equal(meta?.minRuntimeVersion, '0.25.1')
+  assert.equal(meta?.dataCompatibilityVersion, 1)
+  assert.deepEqual(meta?.reasons, ['segment-schema'])
+})
+
+test('checks data directory compatibility before creating a push-import database', async (t) => {
+  const root = makeTempDir()
+  t.after(() => fs.rmSync(root, { recursive: true, force: true }))
+  raiseDataDirMinRuntimeVersion(
+    {
+      getSystemDir: () => root,
+      getUserDataDir: () => root,
+      getDatabaseDir: () => path.join(root, 'databases'),
+      getVectorDir: () => path.join(root, 'vector'),
+      getAiDataDir: () => path.join(root, 'ai'),
+      getSettingsDir: () => path.join(root, 'settings'),
+      getCacheDir: () => path.join(root, 'cache'),
+      getTempDir: () => path.join(root, 'temp'),
+      getLogsDir: () => path.join(root, 'logs'),
+      getDownloadsDir: () => root,
+    },
+    {
+      minRuntimeVersion: '0.26.0',
+      dataCompatibilityVersion: 2,
+      reason: 'future-schema',
+      runtime: { version: '0.26.0', kind: 'desktop' },
+      module: 'future-migration',
+    }
+  )
+  const manager = createDatabaseManager(root, { version: '0.25.1', kind: 'cli' })
+
+  await assert.rejects(
+    () =>
+      pushImport(manager, 'incompatible-data-dir', {
+        chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+        meta: { name: 'Incompatible', platform: 'wechat', type: 'private' },
+        messages: [{ sender: 'wxid_alice', timestamp: 1780330832, type: 0, content: 'hello' }],
+      }),
+    /requires runtime version 0\.26\.0 or newer/
+  )
+  assert.equal(fs.existsSync(path.join(root, 'databases', 'incompatible-data-dir.db')), false)
+})
+
 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 }))
@@ -200,6 +276,76 @@ test('deduplicates incremental push when a later platform-id message matches an
   }
 })
 
+test('keeps distinct platform messages even when timestamp, sender and content match', 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: 'Distinct Platform Messages', 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',
+      },
+    ],
+  }
+
+  const initialOutcome = await pushImport(manager, 'distinct-platform-messages', initialPayload)
+  assert.equal(initialOutcome.ok, true)
+
+  const outcome = await pushImport(manager, 'distinct-platform-messages', {
+    messages: [
+      {
+        platformMessageId: 'msg-2',
+        sender: 'wxid_alice',
+        accountName: 'Alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'same content',
+      },
+    ],
+  })
+
+  assert.equal(outcome.ok, true)
+  if (!outcome.ok) return
+  assert.deepEqual(outcome.result.batch, {
+    receivedCount: 1,
+    writtenCount: 1,
+    duplicateCount: 0,
+  })
+
+  const db = manager.openRawSessionDatabase('distinct-platform-messages', { readonly: true })
+  try {
+    const row = db.prepare('SELECT COUNT(*) as count FROM message').get() as { count: number }
+    assert.equal(row.count, 2)
+  } finally {
+    db.close()
+  }
+})
+
+test('creates push-import databases only inside the canonical databases directory', async (t) => {
+  const tempDir = makeTempDir()
+  t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
+
+  const manager = createDatabaseManager(tempDir)
+  const outcome = await pushImport(manager, 'canonical-path', {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Canonical Path', platform: 'wechat', type: 'private' },
+    messages: [{ sender: 'wxid_alice', timestamp: 1780330832, type: 0, content: 'hello' }],
+  })
+
+  assert.equal(outcome.ok, true)
+  assert.equal(fs.existsSync(path.join(tempDir, 'databases', 'canonical-path.db')), true)
+  assert.equal(fs.existsSync(path.join(tempDir, 'canonical-path.db')), false)
+})
+
 test('rejects non-array messages before applying incremental metadata updates', async (t) => {
   const tempDir = makeTempDir()
   t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))

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

@@ -2,22 +2,25 @@
  * Push import service — handles POST /api/v1/imports/:sessionId
  *
  * Accepts a ChatLab Format JSON payload, creates or appends to a session.
- * Dedup: platformMessageId (preferred) or content hash (fallback).
+ * Dedup: platformMessageId (preferred) or deterministic fallback key.
  */
 
 import * as fs from 'fs'
 import { DataDirCompatibilityError } from '../data-dir-compat'
-import {
-  CHAT_DB_SCHEMA,
-  generateMessageKey,
-  generateSessionIndex,
-  generateIncrementalSessionIndex,
-} from '@openchatlab/core'
+import { CHAT_DB_SCHEMA, generateSessionIndex, generateIncrementalSessionIndex } from '@openchatlab/core'
 import type { DatabaseAdapter } from '@openchatlab/core'
 import type { DatabaseManager } from '../database-manager'
 import { buildFtsIndex, insertFtsEntries } from '../fts'
 import { writeParseResultToDb } from '../import'
 import { ImportInProgressError, withDataDirImportLock } from '../import/import-lock'
+import {
+  createMessageDedupState,
+  generateFallbackMessageKey,
+  registerMessageAndCheckDuplicate,
+  type MessageDedupState,
+} from '../import/message-deduplicator'
+import { appLogger } from '../logging/app-logger'
+import { isValidImportSessionId } from '../import/session-id'
 
 const SYSTEM_SENDER_ID = 'SYSTEM'
 const SYSTEM_MEMBER_NAME = '系统消息'
@@ -73,6 +76,12 @@ export type PushImportOutcome =
   | { ok: true; result: PushImportResult }
   | { ok: false; reason: 'import_in_progress' | 'invalid_payload' | 'import_failed'; message: string }
 
+export interface PushImportExecutionDeps {
+  getDbPath(sessionId: string): string
+  openDatabase(sessionId: string, options: { readonly?: boolean; create?: boolean }): DatabaseAdapter
+  deleteDatabase(sessionId: string): void
+}
+
 function validatePayload(payload: PushImportPayload, isNew: boolean): string | null {
   const messages = payload.messages
   if (!Array.isArray(messages) || messages.length === 0) return 'messages is required and must be a non-empty array'
@@ -142,8 +151,7 @@ function normalizeSenderGroupNickname(sender: string, groupNickname: string | un
 function writeMessages(
   db: DatabaseAdapter,
   messages: PushImportMessage[],
-  existingPmids: Set<string>,
-  existingKeys: Set<string>
+  dedupState: MessageDedupState
 ): {
   writtenCount: number
   duplicateCount: number
@@ -167,26 +175,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(key)
-      } else {
-        if (existingKeys.has(key)) {
-          duplicateCount++
-          continue
-        }
-        existingKeys.add(key)
+      if (
+        registerMessageAndCheckDuplicate(
+          {
+            platformMessageId: msg.platformMessageId,
+            timestamp: msg.timestamp,
+            senderPlatformId: msg.sender,
+            type: msg.type,
+            content: msg.content ?? null,
+            replyToMessageId: msg.replyToMessageId,
+          },
+          dedupState
+        )
+      ) {
+        duplicateCount++
+        continue
       }
 
       let memberId = memberIdCache.get(msg.sender)
@@ -244,29 +247,25 @@ function fullImport(
   const allMembers = extraMembers.length > 0 ? [...members, ...extraMembers] : members
 
   // Deduplicate within the batch using the same pmid/hash logic as incremental imports.
-  const pmids = new Set<string>()
-  const hashKeys = new Set<string>()
+  const dedupState = createMessageDedupState()
   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 {
-      if (hashKeys.has(key)) {
-        duplicateCount++
-        continue
-      }
-      hashKeys.add(key)
+    if (
+      registerMessageAndCheckDuplicate(
+        {
+          platformMessageId: msg.platformMessageId,
+          timestamp: msg.timestamp,
+          senderPlatformId: msg.sender,
+          type: msg.type,
+          content: msg.content ?? null,
+          replyToMessageId: msg.replyToMessageId,
+        },
+        dedupState
+      )
+    ) {
+      duplicateCount++
+      continue
     }
     dedupedMessages.push(msg)
   }
@@ -320,19 +319,34 @@ function incrementalImport(
   // Load dedup keys
   const existingPmids = new Set<string>()
   const existingKeys = new Set<string>()
-  ;(
-    db.prepare('SELECT platform_message_id FROM message WHERE platform_message_id IS NOT NULL').all() as Array<{
-      platform_message_id: string
-    }>
-  ).forEach((r) => existingPmids.add(r.platform_message_id))
-  // Load hashes for ALL existing messages (not just those without pmid) so that
-  // messages previously imported with a platformMessageId are still caught by
-  // content-hash dedup when the same content arrives without one.
-  ;(
-    db
-      .prepare(`SELECT msg.ts, m.platform_id, msg.content FROM message msg JOIN member m ON msg.sender_id = m.id`)
-      .all() as Array<{ ts: number; platform_id: string; content: string | null }>
-  ).forEach((r) => existingKeys.add(generateMessageKey(r.ts, r.platform_id, r.content)))
+  const existingFallbackOnlyKeys = new Set<string>()
+  const existingMessages = db
+    .prepare(
+      `SELECT msg.ts, m.platform_id, msg.type, msg.content, msg.reply_to_message_id, msg.platform_message_id
+       FROM message msg
+       JOIN member m ON msg.sender_id = m.id`
+    )
+    .all() as Array<{
+    ts: number
+    platform_id: string
+    type: number
+    content: string | null
+    reply_to_message_id: string | null
+    platform_message_id: string | null
+  }>
+  for (const message of existingMessages) {
+    if (message.platform_message_id) existingPmids.add(message.platform_message_id)
+    const key = generateFallbackMessageKey({
+      timestamp: message.ts,
+      senderPlatformId: message.platform_id,
+      type: message.type,
+      content: message.content,
+      replyToMessageId: message.reply_to_message_id ?? undefined,
+    })
+    existingKeys.add(key)
+    if (!message.platform_message_id) existingFallbackOnlyKeys.add(key)
+  }
+  const dedupState = createMessageDedupState(existingPmids, existingKeys, existingFallbackOnlyKeys)
 
   let metaUpdated = false
   let membersAdded = 0
@@ -387,12 +401,7 @@ function incrementalImport(
   const preWriteMaxTs =
     (db.prepare('SELECT MAX(ts) as max_ts FROM message').get() as { max_ts: number | null })?.max_ts ?? 0
 
-  const { writtenCount, duplicateCount, minWrittenTs, ftsEntries } = writeMessages(
-    db,
-    payload.messages!,
-    existingPmids,
-    existingKeys
-  )
+  const { writtenCount, duplicateCount, minWrittenTs, ftsEntries } = writeMessages(db, payload.messages!, dedupState)
 
   if (ftsEntries.length > 0) {
     try {
@@ -423,23 +432,34 @@ function incrementalImport(
   return { writtenCount, duplicateCount, metaUpdated, membersAdded, membersUpdated }
 }
 
-// Only allow characters that are safe as a bare filename component.
-// Rejects path separators (/ \), dots-only sequences (..), and control chars.
-const SAFE_SESSION_ID_RE = /^[a-zA-Z0-9_@-][a-zA-Z0-9_@.-]*$/
-
 export async function pushImport(
   dbManager: DatabaseManager,
   sessionId: string,
   payload: PushImportPayload
 ): Promise<PushImportOutcome> {
-  if (!SAFE_SESSION_ID_RE.test(sessionId) || sessionId.includes('..')) {
-    return { ok: false, reason: 'invalid_payload', message: 'sessionId contains invalid characters' }
-  }
-
   try {
-    return await withDataDirImportLock(dbManager.getUserDataDir(), () =>
-      pushImportUnlocked(dbManager, sessionId, payload)
-    )
+    return await withDataDirImportLock(dbManager.getUserDataDir(), async () => {
+      const outcome = await executePushImportUnlocked(
+        {
+          getDbPath: (id) => dbManager.getDbPath(id),
+          openDatabase: (id, options) => dbManager.openRawSessionDatabase(id, options),
+          deleteDatabase: (id) => {
+            dbManager.deleteSessionDatabaseFiles(id)
+          },
+        },
+        sessionId,
+        payload
+      )
+      if (!outcome.ok) return outcome
+
+      try {
+        dbManager.raiseCurrentChatDbCompatibilityGate()
+      } catch (error) {
+        if (outcome.result.created) dbManager.deleteSessionDatabaseFiles(sessionId)
+        throw error
+      }
+      return outcome
+    })
   } catch (error) {
     if (error instanceof ImportInProgressError) {
       return {
@@ -452,12 +472,20 @@ export async function pushImport(
   }
 }
 
-async function pushImportUnlocked(
-  dbManager: DatabaseManager,
+/**
+ * 执行已加锁的 Push Import 写库流程。调用方必须负责数据目录导入锁和兼容门禁，
+ * 这样 CLI/Internal 可使用 DatabaseManager，Desktop 则可安全地在 worker 中执行写库。
+ */
+export async function executePushImportUnlocked(
+  deps: PushImportExecutionDeps,
   sessionId: string,
   payload: PushImportPayload
 ): Promise<PushImportOutcome> {
-  const dbPath = dbManager.getDbPath(sessionId)
+  if (!isValidImportSessionId(sessionId)) {
+    return { ok: false, reason: 'invalid_payload', message: 'sessionId contains invalid characters' }
+  }
+
+  const dbPath = deps.getDbPath(sessionId)
   const isNew = !fs.existsSync(dbPath)
 
   const validationError = validatePayload(payload, isNew)
@@ -467,7 +495,7 @@ async function pushImportUnlocked(
 
   try {
     if (isNew) {
-      const db = dbManager.openRawSessionDatabase(sessionId, { create: true })
+      const db = deps.openDatabase(sessionId, { create: true })
       try {
         const { writtenCount, duplicateCount, membersAdded } = fullImport(
           db,
@@ -476,8 +504,7 @@ async function pushImportUnlocked(
           payload.messages!
         )
         const session = queryStats(db)
-        dbManager.raiseCurrentChatDbCompatibilityGate()
-        return {
+        const outcome: PushImportOutcome = {
           ok: true,
           result: {
             sessionId,
@@ -487,17 +514,18 @@ async function pushImportUnlocked(
             updates: { metaUpdated: true, membersAdded, membersUpdated: 0 },
           },
         }
+        appLogger.info('push-import', 'Push import completed', outcome.result)
+        return outcome
       } finally {
         db.close()
       }
     }
 
-    const db = dbManager.openRawSessionDatabase(sessionId, { readonly: false })
+    const db = deps.openDatabase(sessionId, { readonly: false })
     try {
       const { writtenCount, duplicateCount, metaUpdated, membersAdded, membersUpdated } = incrementalImport(db, payload)
       const session = queryStats(db)
-      dbManager.raiseCurrentChatDbCompatibilityGate()
-      return {
+      const outcome: PushImportOutcome = {
         ok: true,
         result: {
           sessionId,
@@ -507,6 +535,8 @@ async function pushImportUnlocked(
           updates: { metaUpdated, membersAdded, membersUpdated },
         },
       }
+      appLogger.info('push-import', 'Push import completed', outcome.result)
+      return outcome
     } finally {
       db.close()
     }
@@ -517,11 +547,12 @@ async function pushImportUnlocked(
 
     if (isNew) {
       try {
-        dbManager.deleteSessionDatabaseFiles(sessionId)
+        deps.deleteDatabase(sessionId)
       } catch {
         /* cleanup best-effort */
       }
     }
+    appLogger.error('push-import', 'Push import failed', err)
     return { ok: false, reason: 'import_failed', message: err instanceof Error ? err.message : String(err) }
   }
 }