fix(import): align JSON dry-run with push semantics

Route Desktop JSON dry-runs through a shared read-only push analyzer so validation, deduplication, and member inference match actual writes. Keep JSONL on the file analyzer path and cover new and incremental predictions without database mutations.

apps/desktop/main/api/routes/import-helpers.test.ts

@@ -2,7 +2,7 @@ import assert from 'node:assert/strict'
 import test from 'node:test'
 import { IMPORT_IN_PROGRESS_ERROR_KEY } from '@openchatlab/node-runtime/src/import/import-lock'
 import { ApiErrorCode } from '../errors'
-import { analysisFromNewImport, apiErrorFromImportResult } from './import-helpers'
+import { analysisFromNewImport, analysisFromPushImport, apiErrorFromImportResult } from './import-helpers'
 
 test('maps an import lock result to the documented 409 API error', () => {
   const error = apiErrorFromImportResult(IMPORT_IN_PROGRESS_ERROR_KEY, 'Import failed')
@@ -36,3 +36,25 @@ test('maps new-import analysis counts without assuming every parsed message is n
     }
   )
 })
+
+test('preserves the Desktop dry-run response shape for new and existing push imports', () => {
+  const base = {
+    sessionId: 'push-analysis',
+    totalInFile: 3,
+    newMessageCount: 2,
+    duplicateCount: 1,
+    newMemberCount: 4,
+  }
+
+  assert.deepEqual(analysisFromPushImport({ ...base, created: true }), {
+    totalInFile: 3,
+    newMessageCount: 2,
+    duplicateCount: 1,
+    newMemberCount: 4,
+  })
+  assert.deepEqual(analysisFromPushImport({ ...base, created: false }), {
+    totalInFile: 3,
+    newMessageCount: 2,
+    duplicateCount: 1,
+  })
+})

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

@@ -1,4 +1,5 @@
 import { IMPORT_IN_PROGRESS_ERROR_KEY } from '@openchatlab/node-runtime/src/import/import-lock'
+import type { PushImportAnalysisResult } from '@openchatlab/node-runtime/src/services/push-importer'
 import type { AnalyzeNewImportResult } from '@openchatlab/node-runtime/src/import/streaming-importer'
 import { importFailed, importInProgress, type ApiError } from '../errors'
 
@@ -19,3 +20,17 @@ export function analysisFromNewImport(result: AnalyzeNewImportResult): {
     newMemberCount: result.totalMembers,
   }
 }
+
+export function analysisFromPushImport(result: PushImportAnalysisResult): {
+  totalInFile: number
+  newMessageCount: number
+  duplicateCount: number
+  newMemberCount?: number
+} {
+  const analysis = {
+    totalInFile: result.totalInFile,
+    newMessageCount: result.newMessageCount,
+    duplicateCount: result.duplicateCount,
+  }
+  return result.created ? { ...analysis, newMemberCount: result.newMemberCount } : analysis
+}

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

@@ -36,7 +36,7 @@ import {
   errorResponse,
 } from '../errors'
 import { apiLogger } from '../logger'
-import { analysisFromNewImport, apiErrorFromImportResult } from './import-helpers'
+import { analysisFromNewImport, analysisFromPushImport, apiErrorFromImportResult } from './import-helpers'
 
 // Tracks active External API requests; the shared data-directory lock enforces writer exclusion.
 const isImporting = new Set<string>()
@@ -152,9 +152,9 @@ async function handleUnifiedImport(request: FastifyRequest, reply: FastifyReply,
   let idempotencyOwned = false
 
   try {
-    // JSON writes now go straight through the shared push service. JSON dry-runs
-    // and JSONL still need a temp file because their analyzers consume a file path.
-    if (isJsonl || isDryRun) {
+    // JSON writes and dry-runs go straight through the shared push service.
+    // JSONL still needs a temp file because its analyzer and importer consume a file path.
+    if (isJsonl) {
       const writeResult = await writeTempFile(request, isJson)
       if (writeResult.error) {
         const err = invalidFormat(writeResult.error)
@@ -197,6 +197,28 @@ async function handleUnifiedImport(request: FastifyRequest, reply: FastifyReply,
 
     // X-Dry-Run: analyze only, no writes
     if (isDryRun) {
+      if (isJson) {
+        const outcome = await worker.analyzePushImport(sessionId, (request.body ?? {}) as PushImportPayload)
+        if (!outcome.ok) {
+          idempotencyFail(cacheKey)
+          const err =
+            outcome.reason === 'invalid_payload' ? invalidPayload(outcome.message) : importFailed(outcome.message)
+          reply.code(err.statusCode).send(errorResponse(err))
+          return
+        }
+
+        const responsePayload = successResponse({
+          sessionId,
+          created: outcome.result.created,
+          dryRun: true,
+          analysis: analysisFromPushImport(outcome.result),
+        })
+        idempotencySuccess(cacheKey, responsePayload)
+        idempotencyOwned = false
+        reply.send(responsePayload)
+        return
+      }
+
       const exists = sessionExists(sessionId)
       let responsePayload: ImportSuccessResponse
       if (exists) {

apps/desktop/main/worker/dbWorker.ts

@@ -81,6 +81,7 @@ import {
   analyzeIncrementalImport,
   incrementalImport,
   analyzeNewImport,
+  analyzePushImport,
   pushImport,
 } from './import'
 import { initNlpDir } from '@openchatlab/node-runtime'
@@ -249,6 +250,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),
+  analyzePushImport: (p) => analyzePushImport(p.sessionId, p.payload),
   pushImport: (p) => pushImport(p.sessionId, p.payload),
   // 流式解析文件信息（用于合并预览）
   streamParseFileInfo: (p, id) => streamParseFileInfo(p.filePath, id),

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

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

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

@@ -10,10 +10,39 @@ import {
   type MembersCache,
 } from '@openchatlab/node-runtime/src/cache/session-cache'
 import { initDbDir } from '../core/dbCore'
-import { pushImport } from './pushImport'
+import { analyzePushImport, pushImport } from './pushImport'
 
 const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node')
 
+test('Desktop worker push analysis uses shared semantics without creating a database', async (t) => {
+  const baseDir = fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir()
+  const root = fs.mkdtempSync(path.join(baseDir, 'chatlab-worker-push-analysis-'))
+  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 analyzePushImport('worker-analysis', {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Worker Push Analysis', platform: 'wechat', type: 'group' },
+    members: [{ platformId: 'wxid_alice', accountName: 'Alice' }],
+    messages: [{ sender: 'wxid_bob', timestamp: 1780330832, type: 0, content: 'hello' }],
+  })
+
+  assert.deepEqual(outcome, {
+    ok: true,
+    result: {
+      sessionId: 'worker-analysis',
+      created: true,
+      totalInFile: 1,
+      newMessageCount: 1,
+      duplicateCount: 0,
+      newMemberCount: 2,
+    },
+  })
+  assert.equal(fs.existsSync(path.join(dbDir, 'worker-analysis.db')), false)
+})
+
 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-'))

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

@@ -3,7 +3,10 @@ import * as path from 'node:path'
 import { BetterSqliteAdapter } from '@openchatlab/node-runtime/src/better-sqlite3-adapter'
 import { computeAndSetOverviewCache, deleteSessionCache } from '@openchatlab/node-runtime/src/cache/session-cache'
 import {
+  executeAnalyzePushImport,
   executePushImportUnlocked,
+  type PushImportAnalysisOutcome,
+  type PushImportExecutionDeps,
   type PushImportOutcome,
   type PushImportPayload,
 } from '@openchatlab/node-runtime/src/services/push-importer'
@@ -13,6 +16,32 @@ function getDbPath(sessionId: string): string {
   return path.join(getDbDir(), `${sessionId}.db`)
 }
 
+function createPushImportExecutionDeps(): PushImportExecutionDeps {
+  return {
+    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 */
+        }
+      }
+    },
+  }
+}
+
 function refreshImportCaches(sessionId: string): void {
   const cacheDir = getCacheDir()
   try {
@@ -33,34 +62,16 @@ function refreshImportCaches(sessionId: string): void {
 }
 
 export async function pushImport(sessionId: string, payload: PushImportPayload): Promise<PushImportOutcome> {
-  const outcome = await 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
-  )
+  const outcome = await executePushImportUnlocked(createPushImportExecutionDeps(), sessionId, payload)
 
   if (outcome.ok) refreshImportCaches(sessionId)
   return outcome
 }
+
+/** 只读复用 Push Import 的校验、去重和成员推断，不刷新缓存也不创建数据库。 */
+export async function analyzePushImport(
+  sessionId: string,
+  payload: PushImportPayload
+): Promise<PushImportAnalysisOutcome> {
+  return executeAnalyzePushImport(createPushImportExecutionDeps(), sessionId, payload)
+}

apps/desktop/main/worker/workerManager.ts

@@ -21,7 +21,7 @@ import {
   raiseChatDbCompatibilityGate,
   withDataDirImportLock,
 } from '@openchatlab/node-runtime'
-import type { PushImportOutcome, PushImportPayload } from '@openchatlab/node-runtime'
+import type { PushImportAnalysisOutcome, PushImportOutcome, PushImportPayload } from '@openchatlab/node-runtime'
 import { getWorkerRequestTimeoutMs, isRestartableReadOnlyRequestType } from './workerTimeoutPolicy'
 
 interface WorkerRequestOptions {
@@ -650,6 +650,17 @@ export async function pushImport(sessionId: string, payload: PushImportPayload):
   }
 }
 
+/**
+ * 在 worker 中只读分析 JSON Push 请求，确保 Desktop dry-run 与正式写入使用同一套语义。
+ */
+export async function analyzePushImport(
+  sessionId: string,
+  payload: PushImportPayload
+): Promise<PushImportAnalysisOutcome> {
+  assertDataDirCompatibleNow()
+  return sendToWorker<PushImportAnalysisOutcome>('analyzePushImport', { sessionId, payload })
+}
+
 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

@@ -10,6 +10,7 @@ describe('worker timeout policy', () => {
     assert.equal(isRestartableReadOnlyRequestType('getChatOverview'), true)
     assert.equal(isRestartableReadOnlyRequestType('searchMessages'), true)
     assert.equal(isRestartableReadOnlyRequestType('getSegmentSummaries'), true)
+    assert.equal(isRestartableReadOnlyRequestType('analyzePushImport'), true)
   })
 
   it('keeps mutating, indexing, import, export, and unknown requests non-restartable', () => {

apps/desktop/main/worker/workerTimeoutPolicy.ts

@@ -40,6 +40,7 @@ const RESTARTABLE_READ_ONLY_REQUEST_TYPES = new Set([
   'getRecentChatSessions',
   'getSegmentMessages',
   'getSegmentSummaries',
+  'analyzePushImport',
 ])
 
 const DEFAULT_WORKER_REQUEST_TIMEOUT_MS = 60 * 1000

packages/node-runtime/src/index.ts

@@ -428,6 +428,8 @@ export {
   createPeopleRelationshipsService,
   createGlobalInsightService,
   createDatabaseManagerAdapter,
+  analyzePushImport,
+  executeAnalyzePushImport,
   executePushImportUnlocked,
   DEFAULT_IMPORT_IDEMPOTENCY_TTL_MS,
   hashImportBody,
@@ -454,6 +456,9 @@ export type {
   GlobalInsightService,
   GlobalInsightServiceDeps,
   ImportIdempotencyStartResult,
+  PushImportAnalysisExecutionDeps,
+  PushImportAnalysisOutcome,
+  PushImportAnalysisResult,
   PushImportExecutionDeps,
   GlobalInsightServiceOptions,
   PushImportPayload,

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

@@ -81,8 +81,11 @@ export type {
 export { MergeSessionCache } from './merge-cache'
 
 // Push import (POST /api/v1/imports/:sessionId)
-export { executePushImportUnlocked, pushImport } from './push-importer'
+export { analyzePushImport, executeAnalyzePushImport, executePushImportUnlocked, pushImport } from './push-importer'
 export type {
+  PushImportAnalysisExecutionDeps,
+  PushImportAnalysisOutcome,
+  PushImportAnalysisResult,
   PushImportExecutionDeps,
   PushImportPayload,
   PushImportResult,

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

@@ -7,7 +7,7 @@ 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'
+import { analyzePushImport, pushImport } from './push-importer'
 
 const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node')
 
@@ -34,6 +34,105 @@ function createDatabaseManager(rootDir: string, runtime?: RuntimeIdentity): Data
   )
 }
 
+test('analyzes a new push payload with the same optional-account and member semantics as the writer', 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 Analysis', platform: 'wechat', type: 'group' },
+    members: [{ platformId: 'wxid_alice', accountName: 'Alice' }],
+    messages: [{ sender: 'wxid_bob', timestamp: 1780330832, type: 0, content: 'hello' }],
+  }
+
+  const analysis = await analyzePushImport(manager, 'push-analysis', payload)
+
+  assert.deepEqual(analysis, {
+    ok: true,
+    result: {
+      sessionId: 'push-analysis',
+      created: true,
+      totalInFile: 1,
+      newMessageCount: 1,
+      duplicateCount: 0,
+      newMemberCount: 2,
+    },
+  })
+  assert.equal(fs.existsSync(manager.getDbPath('push-analysis')), false)
+
+  const outcome = await pushImport(manager, 'push-analysis', payload)
+  assert.equal(outcome.ok, true)
+  if (!outcome.ok || !analysis.ok) return
+  assert.equal(outcome.result.batch.writtenCount, analysis.result.newMessageCount)
+  assert.equal(outcome.result.batch.duplicateCount, analysis.result.duplicateCount)
+  assert.equal(outcome.result.updates.membersAdded, analysis.result.newMemberCount)
+})
+
+test('analyzes incremental push deduplication and member creation without writing', async (t) => {
+  const tempDir = makeTempDir()
+  t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
+
+  const manager = createDatabaseManager(tempDir)
+  const initialOutcome = await pushImport(manager, 'incremental-analysis', {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Incremental Analysis', platform: 'wechat', type: 'group' },
+    members: [{ platformId: 'wxid_alice', accountName: 'Alice' }],
+    messages: [
+      {
+        platformMessageId: 'msg-1',
+        sender: 'wxid_alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'hello',
+      },
+    ],
+  })
+  assert.equal(initialOutcome.ok, true)
+
+  const payload: PushImportPayload = {
+    members: [{ platformId: 'wxid_charlie', accountName: 'Charlie' }],
+    messages: [
+      {
+        platformMessageId: 'msg-1',
+        sender: 'wxid_alice',
+        timestamp: 1780330832,
+        type: 0,
+        content: 'hello',
+      },
+      { sender: 'wxid_bob', timestamp: 1780330833, type: 0, content: 'new message' },
+    ],
+  }
+
+  const analysis = await analyzePushImport(manager, 'incremental-analysis', payload)
+  assert.deepEqual(analysis, {
+    ok: true,
+    result: {
+      sessionId: 'incremental-analysis',
+      created: false,
+      totalInFile: 2,
+      newMessageCount: 1,
+      duplicateCount: 1,
+      newMemberCount: 2,
+    },
+  })
+
+  const beforeWrite = manager.openRawSessionDatabase('incremental-analysis', { readonly: true })
+  try {
+    assert.equal((beforeWrite.prepare('SELECT COUNT(*) as count FROM message').get() as { count: number }).count, 1)
+    assert.equal((beforeWrite.prepare('SELECT COUNT(*) as count FROM member').get() as { count: number }).count, 1)
+  } finally {
+    beforeWrite.close()
+  }
+
+  const outcome = await pushImport(manager, 'incremental-analysis', payload)
+  assert.equal(outcome.ok, true)
+  if (!outcome.ok || !analysis.ok) return
+  assert.equal(outcome.result.batch.writtenCount, analysis.result.newMessageCount)
+  assert.equal(outcome.result.batch.duplicateCount, analysis.result.duplicateCount)
+  assert.equal(outcome.result.updates.membersAdded, analysis.result.newMemberCount)
+})
+
 test('rejects push imports while any writer holds the data-directory import lock', async (t) => {
   const root = makeTempDir()
   t.after(() => fs.rmSync(root, { recursive: true, force: true }))

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

@@ -76,12 +76,27 @@ export type PushImportOutcome =
   | { ok: true; result: PushImportResult }
   | { ok: false; reason: 'import_in_progress' | 'invalid_payload' | 'import_failed'; message: string }
 
+export interface PushImportAnalysisResult {
+  sessionId: string
+  created: boolean
+  totalInFile: number
+  newMessageCount: number
+  duplicateCount: number
+  newMemberCount: number
+}
+
+export type PushImportAnalysisOutcome =
+  | { ok: true; result: PushImportAnalysisResult }
+  | { ok: false; reason: '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
 }
 
+export type PushImportAnalysisExecutionDeps = Pick<PushImportExecutionDeps, 'getDbPath' | 'openDatabase'>
+
 function isRecord(value: unknown): value is Record<string, unknown> {
   return typeof value === 'object' && value !== null && !Array.isArray(value)
 }
@@ -200,6 +215,77 @@ function isCountedMember(platformId: string): boolean {
   return platformId !== SYSTEM_SENDER_ID
 }
 
+function registerPushMessageAndCheckDuplicate(message: PushImportMessage, dedupState: MessageDedupState): boolean {
+  return registerMessageAndCheckDuplicate(
+    {
+      platformMessageId: message.platformMessageId,
+      timestamp: message.timestamp,
+      senderPlatformId: message.sender,
+      type: message.type,
+      content: message.content ?? null,
+      replyToMessageId: message.replyToMessageId,
+    },
+    dedupState
+  )
+}
+
+function createExistingMessageDedupState(db: DatabaseAdapter): MessageDedupState {
+  const existingPmids = new Set<string>()
+  const existingKeys = new Set<string>()
+  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)
+  }
+
+  return createMessageDedupState(existingPmids, existingKeys, existingFallbackOnlyKeys)
+}
+
+function analyzePushMessages(
+  messages: PushImportMessage[],
+  dedupState: MessageDedupState,
+  sortByTimestamp: boolean
+): { newMessageCount: number; duplicateCount: number; writableSenders: Set<string> } {
+  const orderedMessages = sortByTimestamp ? [...messages].sort((a, b) => a.timestamp - b.timestamp) : messages
+  const writableSenders = new Set<string>()
+  let newMessageCount = 0
+  let duplicateCount = 0
+
+  for (const message of orderedMessages) {
+    if (registerPushMessageAndCheckDuplicate(message, dedupState)) {
+      duplicateCount++
+      continue
+    }
+    newMessageCount++
+    writableSenders.add(message.sender)
+  }
+
+  return { newMessageCount, duplicateCount, writableSenders }
+}
+
 function writeMessages(
   db: DatabaseAdapter,
   messages: PushImportMessage[],
@@ -231,19 +317,7 @@ function writeMessages(
 
   db.transaction(() => {
     for (const msg of sorted) {
-      if (
-        registerMessageAndCheckDuplicate(
-          {
-            platformMessageId: msg.platformMessageId,
-            timestamp: msg.timestamp,
-            senderPlatformId: msg.sender,
-            type: msg.type,
-            content: msg.content ?? null,
-            replyToMessageId: msg.replyToMessageId,
-          },
-          dedupState
-        )
-      ) {
+      if (registerPushMessageAndCheckDuplicate(msg, dedupState)) {
         duplicateCount++
         continue
       }
@@ -312,19 +386,7 @@ function fullImport(
   let duplicateCount = 0
   const dedupedMessages: PushImportMessage[] = []
   for (const msg of messages) {
-    if (
-      registerMessageAndCheckDuplicate(
-        {
-          platformMessageId: msg.platformMessageId,
-          timestamp: msg.timestamp,
-          senderPlatformId: msg.sender,
-          type: msg.type,
-          content: msg.content ?? null,
-          replyToMessageId: msg.replyToMessageId,
-        },
-        dedupState
-      )
-    ) {
+    if (registerPushMessageAndCheckDuplicate(msg, dedupState)) {
       duplicateCount++
       continue
     }
@@ -376,37 +438,7 @@ function writeIncrementalImport(db: DatabaseAdapter, payload: PushImportPayload)
   const metaUpdateMode = payload.options?.metaUpdateMode ?? 'patch'
   const memberUpdateMode = payload.options?.memberUpdateMode ?? 'upsert'
 
-  // Load dedup keys
-  const existingPmids = new Set<string>()
-  const existingKeys = new Set<string>()
-  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)
+  const dedupState = createExistingMessageDedupState(db)
 
   let metaUpdated = false
   let membersAdded = 0
@@ -508,6 +540,119 @@ function incrementalImport(db: DatabaseAdapter, payload: PushImportPayload): Inc
   return db.transaction(() => writeIncrementalImport(db, payload))
 }
 
+function countNewSessionMembers(payload: PushImportPayload): number {
+  const memberIds = new Set<string>()
+  for (const member of payload.members ?? []) {
+    if (isCountedMember(member.platformId)) memberIds.add(member.platformId)
+  }
+  for (const message of payload.messages ?? []) {
+    if (isCountedMember(message.sender)) memberIds.add(message.sender)
+  }
+  return memberIds.size
+}
+
+function countIncrementalMembersAdded(
+  db: DatabaseAdapter,
+  payload: PushImportPayload,
+  writableSenders: Set<string>
+): number {
+  const knownMemberIds = new Set(
+    (db.prepare('SELECT platform_id FROM member').all() as Array<{ platform_id: string }>).map(
+      (member) => member.platform_id
+    )
+  )
+  let membersAdded = 0
+
+  if ((payload.options?.memberUpdateMode ?? 'upsert') === 'upsert') {
+    for (const member of payload.members ?? []) {
+      if (!isCountedMember(member.platformId) || knownMemberIds.has(member.platformId)) continue
+      knownMemberIds.add(member.platformId)
+      membersAdded++
+    }
+  }
+
+  for (const sender of writableSenders) {
+    if (!isCountedMember(sender) || knownMemberIds.has(sender)) continue
+    knownMemberIds.add(sender)
+    membersAdded++
+  }
+
+  return membersAdded
+}
+
+/**
+ * 只读分析 JSON Push 请求。新会话按首次导入的批内去重顺序计算；已有会话则加载
+ * 当前消息去重键，并按增量写入的时间顺序计算，确保 dry-run 与正式写入共享同一语义。
+ */
+export async function executeAnalyzePushImport(
+  deps: PushImportAnalysisExecutionDeps,
+  sessionId: string,
+  payload: PushImportPayload
+): Promise<PushImportAnalysisOutcome> {
+  if (!isValidImportSessionId(sessionId)) {
+    return { ok: false, reason: 'invalid_payload', message: 'sessionId contains invalid characters' }
+  }
+
+  const isNew = !fs.existsSync(deps.getDbPath(sessionId))
+  const validationError = validatePayload(payload, isNew)
+  if (validationError) {
+    return { ok: false, reason: 'invalid_payload', message: validationError }
+  }
+
+  try {
+    let newMessageCount: number
+    let duplicateCount: number
+    let newMemberCount: number
+
+    if (isNew) {
+      const analysis = analyzePushMessages(payload.messages!, createMessageDedupState(), false)
+      newMessageCount = analysis.newMessageCount
+      duplicateCount = analysis.duplicateCount
+      newMemberCount = countNewSessionMembers(payload)
+    } else {
+      const db = deps.openDatabase(sessionId, { readonly: true })
+      try {
+        const analysis = analyzePushMessages(payload.messages!, createExistingMessageDedupState(db), true)
+        newMessageCount = analysis.newMessageCount
+        duplicateCount = analysis.duplicateCount
+        newMemberCount = countIncrementalMembersAdded(db, payload, analysis.writableSenders)
+      } finally {
+        db.close()
+      }
+    }
+
+    const result: PushImportAnalysisResult = {
+      sessionId,
+      created: isNew,
+      totalInFile: payload.messages!.length,
+      newMessageCount,
+      duplicateCount,
+      newMemberCount,
+    }
+    appLogger.info('push-import', 'Push import analysis completed', result)
+    return { ok: true, result }
+  } catch (err: unknown) {
+    if (err instanceof DataDirCompatibilityError) throw err
+    appLogger.error('push-import', 'Push import analysis failed', err)
+    return { ok: false, reason: 'import_failed', message: err instanceof Error ? err.message : String(err) }
+  }
+}
+
+export async function analyzePushImport(
+  dbManager: DatabaseManager,
+  sessionId: string,
+  payload: PushImportPayload
+): Promise<PushImportAnalysisOutcome> {
+  return executeAnalyzePushImport(
+    {
+      getDbPath: (id) => dbManager.getDbPath(id),
+      openDatabase: (id, options) => dbManager.openRawSessionDatabase(id, options),
+    },
+    sessionId,
+    payload
+  )
+}
+
 export async function pushImport(
   dbManager: DatabaseManager,
   sessionId: string,