perf(import): avoid loading existing messages for dedup

packages/node-runtime/src/import/incremental-importer.test.ts

@@ -93,6 +93,53 @@ function writeChatLabJson(filePath: string): void {
   )
 }
 
+interface DedupFixtureMessage {
+  timestamp?: number
+  type?: number
+  content?: string | null
+  replyToMessageId?: string
+  platformMessageId?: string
+}
+
+function writeDedupFixture(filePath: string, messages: DedupFixtureMessage[]): void {
+  fs.writeFileSync(
+    filePath,
+    JSON.stringify({
+      chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+      meta: { name: 'Dedup Fixture', platform: 'wechat', type: 'private' },
+      members: [{ platformId: 'wxid_alice', accountName: 'Alice' }],
+      messages: messages.map((message) => ({
+        sender: 'wxid_alice',
+        accountName: 'Alice',
+        timestamp: message.timestamp ?? 1780330832,
+        type: message.type ?? 0,
+        content: Object.hasOwn(message, 'content') ? message.content : 'same message',
+        replyToMessageId: message.replyToMessageId,
+        platformMessageId: message.platformMessageId,
+      })),
+    }),
+    'utf8'
+  )
+}
+
+function seedExistingMessage(dbPath: string, message: DedupFixtureMessage): void {
+  const db = openBetterSqliteDatabase(dbPath, { nativeBinding })
+  db.prepare('INSERT INTO member (platform_id, account_name) VALUES (?, ?)').run('wxid_alice', 'Alice')
+  const member = db.prepare('SELECT id FROM member WHERE platform_id = ?').get('wxid_alice') as { id: number }
+  db.prepare(
+    `INSERT INTO message (sender_id, ts, type, content, reply_to_message_id, platform_message_id)
+     VALUES (?, ?, ?, ?, ?, ?)`
+  ).run(
+    member.id,
+    message.timestamp ?? 1780330832,
+    message.type ?? 0,
+    Object.hasOwn(message, 'content') ? (message.content ?? null) : 'same message',
+    message.replyToMessageId ?? null,
+    message.platformMessageId ?? null
+  )
+  db.close()
+}
+
 function seedSessionDb(dbPath: string): void {
   const db = openBetterSqliteDatabase(dbPath, { nativeBinding })
   db.exec(CHAT_DB_SCHEMA)
@@ -355,3 +402,159 @@ test('deduplicates an ID-bearing copy of an existing fallback-only message', asy
   readonlyDb.close()
   assert.equal(row.count, 1)
 })
+
+test('preserves canonical dedup semantics while checking existing database candidates', async (t) => {
+  const scenarios: Array<{
+    name: string
+    existing: DedupFixtureMessage
+    incoming: DedupFixtureMessage[]
+    expectedNew: number
+    expectedDuplicates: number
+    expectedPlatformIds: Array<string | null>
+  }> = [
+    {
+      name: 'same platform ID wins even when content changed',
+      existing: { platformMessageId: 'msg-1', content: 'original' },
+      incoming: [{ platformMessageId: 'msg-1', content: 'edited' }],
+      expectedNew: 0,
+      expectedDuplicates: 1,
+      expectedPlatformIds: ['msg-1'],
+    },
+    {
+      name: 'fallback-only input matches an existing ID-bearing message',
+      existing: { platformMessageId: 'msg-1' },
+      incoming: [{}],
+      expectedNew: 0,
+      expectedDuplicates: 1,
+      expectedPlatformIds: ['msg-1'],
+    },
+    {
+      name: 'different stable IDs survive an identical fallback fingerprint',
+      existing: { platformMessageId: 'msg-1' },
+      incoming: [{ platformMessageId: 'msg-2' }],
+      expectedNew: 1,
+      expectedDuplicates: 0,
+      expectedPlatformIds: ['msg-1', 'msg-2'],
+    },
+    {
+      name: 'one fallback-only row bridges only one stable ID',
+      existing: {},
+      incoming: [{ platformMessageId: 'msg-1' }, { platformMessageId: 'msg-2' }],
+      expectedNew: 1,
+      expectedDuplicates: 1,
+      expectedPlatformIds: [null, 'msg-2'],
+    },
+    {
+      name: 'a bridged stable ID remains duplicate when repeated',
+      existing: {},
+      incoming: [{ platformMessageId: 'msg-1' }, { platformMessageId: 'msg-1' }, { platformMessageId: 'msg-2' }],
+      expectedNew: 1,
+      expectedDuplicates: 2,
+      expectedPlatformIds: [null, 'msg-2'],
+    },
+    {
+      name: 'an empty stored platform ID behaves as fallback-only',
+      existing: { platformMessageId: '' },
+      incoming: [{ platformMessageId: 'msg-1' }],
+      expectedNew: 0,
+      expectedDuplicates: 1,
+      expectedPlatformIds: [''],
+    },
+    {
+      name: 'empty and null content share the canonical fallback fingerprint',
+      existing: { content: null },
+      incoming: [{ content: '' }],
+      expectedNew: 0,
+      expectedDuplicates: 1,
+      expectedPlatformIds: [null],
+    },
+    {
+      name: 'reply target remains part of the fallback fingerprint',
+      existing: { replyToMessageId: 'reply-1' },
+      incoming: [{ replyToMessageId: 'reply-2' }],
+      expectedNew: 1,
+      expectedDuplicates: 0,
+      expectedPlatformIds: [null, null],
+    },
+  ]
+
+  for (const scenario of scenarios) {
+    await t.test(scenario.name, async (st) => {
+      const tempDir = makeTempDir()
+      st.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
+
+      const dbPath = path.join(tempDir, 'session.db')
+      const filePath = path.join(tempDir, 'incoming.json')
+      seedSessionDb(dbPath)
+      seedExistingMessage(dbPath, scenario.existing)
+      writeDedupFixture(filePath, scenario.incoming)
+
+      const deps = createDeps(dbPath)
+      const analysis = await analyzeIncrementalImport('session', filePath, deps)
+      assert.equal(analysis.newMessageCount, scenario.expectedNew)
+      assert.equal(analysis.duplicateCount, scenario.expectedDuplicates)
+
+      const result = await incrementalImport('session', filePath, deps)
+      assert.equal(result.success, true)
+      assert.equal(result.newMessageCount, scenario.expectedNew)
+      assert.equal(result.batch?.duplicateCount, scenario.expectedDuplicates)
+
+      const db = openBetterSqliteDatabase(dbPath, { readonly: true, nativeBinding })
+      const rows = db.prepare('SELECT platform_message_id FROM message ORDER BY id').all() as Array<{
+        platform_message_id: string | null
+      }>
+      db.close()
+      assert.deepEqual(
+        rows.map((row) => row.platform_message_id),
+        scenario.expectedPlatformIds
+      )
+    })
+  }
+})
+
+test('deduplicates repeated input across parser message batches', async (t) => {
+  const tempDir = makeTempDir()
+  t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
+
+  const dbPath = path.join(tempDir, 'session.db')
+  const filePath = path.join(tempDir, 'cross-batch.jsonl')
+  seedSessionDb(dbPath)
+
+  const lines: unknown[] = [
+    {
+      _type: 'header',
+      chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+      meta: { name: 'Cross Batch', platform: 'wechat', type: 'private' },
+    },
+    { _type: 'member', platformId: 'wxid_alice', accountName: 'Alice' },
+  ]
+  for (let index = 0; index < 5000; index++) {
+    lines.push({
+      _type: 'message',
+      sender: 'wxid_alice',
+      timestamp: 1780330832 + index,
+      type: 0,
+      content: `message-${index}`,
+      platformMessageId: `msg-${index}`,
+    })
+  }
+  lines.push({
+    _type: 'message',
+    sender: 'wxid_alice',
+    timestamp: 1780330832,
+    type: 0,
+    content: 'message-0',
+    platformMessageId: 'msg-0',
+  })
+  fs.writeFileSync(filePath, `${lines.map((line) => JSON.stringify(line)).join('\n')}\n`, 'utf8')
+
+  const deps = createDeps(dbPath)
+  const analysis = await analyzeIncrementalImport('session', filePath, deps)
+  assert.equal(analysis.newMessageCount, 5000)
+  assert.equal(analysis.duplicateCount, 1)
+
+  const result = await incrementalImport('session', filePath, deps)
+  assert.equal(result.success, true)
+  assert.equal(result.newMessageCount, 5000)
+  assert.equal(result.batch?.duplicateCount, 1)
+})

packages/node-runtime/src/import/incremental-importer.ts

@@ -8,10 +8,22 @@
  * Callers provide DatabaseAdapter + progress callback via dependency injection.
  */
 
-import type { DatabaseAdapter } from '@openchatlab/core'
+import type { DatabaseAdapter, PreparedStatement } from '@openchatlab/core'
 import { generateSessionIndex, generateIncrementalSessionIndex, getSessionIndexStats } from '@openchatlab/core'
-import { streamParseFile, detectFormat, getFormatFeatureById, type ParseProgress } from '@openchatlab/parser'
-import { generateFallbackMessageKey, registerMessageAndCheckDuplicate } from './message-deduplicator'
+import {
+  streamParseFile,
+  detectFormat,
+  getFormatFeatureById,
+  type ParseProgress,
+  type StreamParseCallbacks,
+} from '@openchatlab/parser'
+import {
+  createMessageDedupState,
+  generateFallbackMessageKey,
+  registerMessageAndCheckDuplicate,
+  type DedupMessage,
+  type MessageDedupState,
+} from './message-deduplicator'
 import { normalizeSystemMemberName, SYSTEM_MEMBER_NAME, type ImportProgressCallback } from './streaming-importer'
 
 // ==================== Public interfaces ====================
@@ -66,76 +78,159 @@ export interface IncrementalImportDeps {
   /** Open existing session DB for read-only access (analyze) or read-write (import). */
   openDatabase(sessionId: string, readonly?: boolean): DatabaseAdapter
   onProgress: ImportProgressCallback
+  /** Optional parser diagnostics hook (e.g. native parser fallback monitoring). */
+  onParserLog?: NonNullable<StreamParseCallbacks['onLog']>
   /** Optional hook after incremental import (e.g. update overview cache). */
   postImportHook?: (db: DatabaseAdapter, sessionId: string) => void | Promise<void>
 }
 
 // ==================== Internal helpers ====================
 
-function loadExistingDedup(db: DatabaseAdapter): {
-  existingPlatformMsgIds: Set<string>
-  existingKeys: Set<string>
-  existingFallbackOnlyKeys: Set<string>
-} {
-  const existingPlatformMsgIds = new Set<string>()
-  const existingKeys = new Set<string>()
-  const existingFallbackOnlyKeys = new Set<string>()
+const PLATFORM_ID_QUERY_CHUNK_SIZE = 500
 
-  const pmidRows = db
-    .prepare('SELECT platform_message_id FROM message WHERE platform_message_id IS NOT NULL')
-    .all() as Array<{ platform_message_id: string }>
-  for (const row of pmidRows) {
-    existingPlatformMsgIds.add(row.platform_message_id)
+interface ExistingMessageCandidate {
+  type: number
+  content: string | null
+  reply_to_message_id: string | null
+  platform_message_id: string | null
+}
+
+interface ExistingMessageLookup {
+  db: DatabaseAdapter
+  maxMessageId: number
+  hasFallbackOnlyMessages: boolean
+  consumedFallbackOnlyKeys: Set<string>
+  bridgedPlatformMessageIds: Set<string>
+  candidateStatement: PreparedStatement
+  platformIdStatements: Map<number, PreparedStatement>
+}
+
+type CandidateCache = Map<string, ExistingMessageCandidate[]>
+
+function createExistingMessageLookup(db: DatabaseAdapter): ExistingMessageLookup {
+  const maxMessageId =
+    (db.prepare('SELECT MAX(id) AS max_id FROM message').get() as { max_id: number | null } | undefined)?.max_id ?? 0
+  const hasFallbackOnlyMessages = Boolean(
+    db
+      .prepare(
+        `SELECT 1 FROM message
+         WHERE id <= ? AND (platform_message_id IS NULL OR platform_message_id = '')
+         LIMIT 1`
+      )
+      .get(maxMessageId)
+  )
+
+  return {
+    db,
+    maxMessageId,
+    hasFallbackOnlyMessages,
+    consumedFallbackOnlyKeys: new Set<string>(),
+    bridgedPlatformMessageIds: new Set<string>(),
+    candidateStatement: db.prepare(
+      `SELECT msg.type, msg.content, msg.reply_to_message_id, msg.platform_message_id
+       FROM member m
+       JOIN message msg ON msg.sender_id = m.id
+       WHERE m.platform_id = ? AND msg.ts = ? AND msg.id <= ?`
+    ),
+    platformIdStatements: new Map<number, PreparedStatement>(),
+  }
+}
+
+function findExistingPlatformMessageIds(
+  lookup: ExistingMessageLookup,
+  platformMessageIds: Iterable<string | undefined>
+): Set<string> {
+  const uniqueIds = [...new Set([...platformMessageIds].filter((id): id is string => Boolean(id)))]
+  const existingIds = new Set<string>()
+
+  for (let start = 0; start < uniqueIds.length; start += PLATFORM_ID_QUERY_CHUNK_SIZE) {
+    const chunk = uniqueIds.slice(start, start + PLATFORM_ID_QUERY_CHUNK_SIZE)
+    let statement = lookup.platformIdStatements.get(chunk.length)
+    if (!statement) {
+      const placeholders = chunk.map(() => '?').join(', ')
+      statement = lookup.db.prepare(
+        `SELECT platform_message_id
+         FROM message
+         WHERE id <= ? AND platform_message_id IN (${placeholders})`
+      )
+      lookup.platformIdStatements.set(chunk.length, statement)
+    }
+    const rows = statement.all(lookup.maxMessageId, ...chunk) as Array<{ platform_message_id: string }>
+    for (const row of rows) existingIds.add(row.platform_message_id)
   }
 
-  const hashRows = db
-    .prepare(
-      `SELECT ts, m.platform_id as sender_platform_id, msg.type, content, reply_to_message_id,
-              platform_message_id
-       FROM message msg
-       JOIN member m ON msg.sender_id = m.id`
+  return existingIds
+}
+
+function getExistingCandidates(
+  lookup: ExistingMessageLookup,
+  message: DedupMessage,
+  cache: CandidateCache
+): ExistingMessageCandidate[] {
+  const cacheKey = JSON.stringify([message.senderPlatformId, message.timestamp])
+  const cached = cache.get(cacheKey)
+  if (cached) return cached
+
+  const rows = lookup.candidateStatement.all(
+    message.senderPlatformId,
+    message.timestamp,
+    lookup.maxMessageId
+  ) as unknown as ExistingMessageCandidate[]
+  cache.set(cacheKey, rows)
+  return rows
+}
+
+function candidateFallbackKey(message: DedupMessage, candidate: ExistingMessageCandidate): string {
+  return generateFallbackMessageKey({
+    timestamp: message.timestamp,
+    senderPlatformId: message.senderPlatformId,
+    type: candidate.type,
+    content: candidate.content,
+    replyToMessageId: candidate.reply_to_message_id ?? undefined,
+  })
+}
+
+function isExistingMessageDuplicate(
+  lookup: ExistingMessageLookup,
+  message: DedupMessage,
+  existingBatchPlatformIds: Set<string>,
+  candidateCache: CandidateCache
+): boolean {
+  if (
+    message.platformMessageId &&
+    (existingBatchPlatformIds.has(message.platformMessageId) ||
+      lookup.bridgedPlatformMessageIds.has(message.platformMessageId))
+  ) {
+    return true
+  }
+
+  const fallbackKey = generateFallbackMessageKey(message)
+  if (message.platformMessageId) {
+    if (!lookup.hasFallbackOnlyMessages || lookup.consumedFallbackOnlyKeys.has(fallbackKey)) return false
+    const matchesFallbackOnly = getExistingCandidates(lookup, message, candidateCache).some(
+      (candidate) => !candidate.platform_message_id && candidateFallbackKey(message, candidate) === fallbackKey
     )
-    .all() as Array<{
-    ts: number
-    sender_platform_id: string
-    type: number
-    content: string | null
-    reply_to_message_id: string | null
-    platform_message_id: string | null
-  }>
-  for (const row of hashRows) {
-    const key = generateFallbackMessageKey({
-      timestamp: row.ts,
-      senderPlatformId: row.sender_platform_id,
-      type: row.type,
-      content: row.content,
-      replyToMessageId: row.reply_to_message_id ?? undefined,
-    })
-    existingKeys.add(key)
-    if (!row.platform_message_id) existingFallbackOnlyKeys.add(key)
+    if (matchesFallbackOnly) {
+      lookup.consumedFallbackOnlyKeys.add(fallbackKey)
+      lookup.bridgedPlatformMessageIds.add(message.platformMessageId)
+    }
+    return matchesFallbackOnly
   }
 
-  return { existingPlatformMsgIds, existingKeys, existingFallbackOnlyKeys }
+  return getExistingCandidates(lookup, message, candidateCache).some(
+    (candidate) => candidateFallbackKey(message, candidate) === fallbackKey
+  )
 }
 
 function isDuplicate(
-  msg: {
-    platformMessageId?: string
-    timestamp: number
-    senderPlatformId: string
-    type: number
-    content: string | null
-    replyToMessageId?: string
-  },
-  existingPlatformMsgIds: Set<string>,
-  existingKeys: Set<string>,
-  existingFallbackOnlyKeys: Set<string>
+  lookup: ExistingMessageLookup,
+  incomingState: MessageDedupState,
+  message: DedupMessage,
+  existingBatchPlatformIds: Set<string>,
+  candidateCache: CandidateCache
 ): boolean {
-  return registerMessageAndCheckDuplicate(msg, {
-    platformMessageIds: existingPlatformMsgIds,
-    fallbackKeys: existingKeys,
-    fallbackOnlyKeys: existingFallbackOnlyKeys,
-  })
+  if (isExistingMessageDuplicate(lookup, message, existingBatchPlatformIds, candidateCache)) return true
+  return registerMessageAndCheckDuplicate(message, incomingState)
 }
 
 export function normalizeImportTimestamp(timestamp: unknown): number | null {
@@ -163,41 +258,60 @@ export async function analyzeIncrementalImport(
     return { error: 'error.session_not_found', newMessageCount: 0, duplicateCount: 0, totalInFile: 0 }
   }
 
-  const { existingPlatformMsgIds, existingKeys, existingFallbackOnlyKeys } = loadExistingDedup(db)
-  db.close()
-
   let totalInFile = 0
   let newMessageCount = 0
   let duplicateCount = 0
   let platform = formatFeature.platform
 
-  await streamParseFile(
-    filePath,
-    {
-      formatOptions: options?.chatIndex === undefined ? undefined : { chatIndex: options.chatIndex },
-      onMeta: (meta) => {
-        platform = meta.platform
-      },
-      onMembers: () => {},
-      onProgress: (progress: ParseProgress) => {
-        deps.onProgress(progress)
-      },
-      onMessageBatch: (batch) => {
-        for (const msg of batch) {
-          totalInFile++
-          const timestamp = normalizeImportTimestamp(msg.timestamp)
-          if (timestamp === null) continue
+  try {
+    const existingLookup = createExistingMessageLookup(db)
+    const incomingState = createMessageDedupState()
 
-          if (isDuplicate({ ...msg, timestamp }, existingPlatformMsgIds, existingKeys, existingFallbackOnlyKeys)) {
-            duplicateCount++
-          } else {
-            newMessageCount++
+    await streamParseFile(
+      filePath,
+      {
+        formatOptions: options?.chatIndex === undefined ? undefined : { chatIndex: options.chatIndex },
+        onMeta: (meta) => {
+          platform = meta.platform
+        },
+        onMembers: () => {},
+        onProgress: (progress: ParseProgress) => {
+          deps.onProgress(progress)
+        },
+        onLog: deps.onParserLog,
+        onMessageBatch: (batch) => {
+          const existingBatchPlatformIds = findExistingPlatformMessageIds(
+            existingLookup,
+            batch.map((message) => message.platformMessageId)
+          )
+          const candidateCache: CandidateCache = new Map()
+
+          for (const msg of batch) {
+            totalInFile++
+            const timestamp = normalizeImportTimestamp(msg.timestamp)
+            if (timestamp === null) continue
+
+            if (
+              isDuplicate(
+                existingLookup,
+                incomingState,
+                { ...msg, timestamp },
+                existingBatchPlatformIds,
+                candidateCache
+              )
+            ) {
+              duplicateCount++
+            } else {
+              newMessageCount++
+            }
           }
-        }
+        },
       },
-    },
-    options?.formatId
-  )
+      options?.formatId
+    )
+  } finally {
+    db.close()
+  }
 
   return { newMessageCount, duplicateCount, totalInFile, platform }
 }
@@ -226,7 +340,8 @@ export async function incrementalImport(
   const memberUpdateMode = options?.memberUpdateMode ?? 'upsert'
 
   try {
-    const { existingPlatformMsgIds, existingKeys, existingFallbackOnlyKeys } = loadExistingDedup(db)
+    const existingLookup = createExistingMessageLookup(db)
+    const incomingState = createMessageDedupState()
 
     const memberIdMap = new Map<string, number>()
     const existingMembers = db.prepare('SELECT id, platform_id FROM member').all() as Array<{
@@ -336,7 +451,14 @@ export async function incrementalImport(
         onProgress: (progress: ParseProgress) => {
           deps.onProgress(progress)
         },
+        onLog: deps.onParserLog,
         onMessageBatch: (batch) => {
+          const existingBatchPlatformIds = findExistingPlatformMessageIds(
+            existingLookup,
+            batch.map((message) => message.platformMessageId)
+          )
+          const candidateCache: CandidateCache = new Map()
+
           for (const msg of batch) {
             processedCount++
 
@@ -354,7 +476,15 @@ export async function incrementalImport(
               continue
             }
 
-            if (isDuplicate({ ...msg, timestamp }, existingPlatformMsgIds, existingKeys, existingFallbackOnlyKeys)) {
+            if (
+              isDuplicate(
+                existingLookup,
+                incomingState,
+                { ...msg, timestamp },
+                existingBatchPlatformIds,
+                candidateCache
+              )
+            ) {
               duplicateCount++
               continue
             }

scripts/bench-streaming-import.mts

@@ -5,7 +5,7 @@
  * Fixture generation and result verification are excluded from the timed range.
  *
  * Usage:
- *   pnpm exec tsx scripts/bench-streaming-import.mts single 100000 [runs=3]
+ *   pnpm exec tsx scripts/bench-streaming-import.mts single 100000 [runs=3] [delta|full]
  *   pnpm exec tsx scripts/bench-streaming-import.mts batch 100 10000 [runs=3] [concurrency=1]
  *
  * Single-file runs also append 1,000 messages incrementally and measure a
@@ -24,7 +24,11 @@ import { getNativeParserStatus, isNativeFormatAvailable, type NativeParserStatus
 import { BetterSqliteAdapter } from '../packages/node-runtime/src/better-sqlite3-adapter'
 import { computeAndSetOverviewCache } from '../packages/node-runtime/src/cache/session-cache'
 import { runKeyedBatch } from '../packages/node-runtime/src/import/batch-coordinator'
-import { incrementalImport, type IncrementalImportDeps } from '../packages/node-runtime/src/import/incremental-importer'
+import {
+  analyzeIncrementalImport,
+  incrementalImport,
+  type IncrementalImportDeps,
+} from '../packages/node-runtime/src/import/incremental-importer'
 import {
   streamingImport,
   type ImportLogger,
@@ -59,8 +63,14 @@ interface BenchmarkResult {
   inputMessages: number
   inputBytes: number
   durationMs: number
+  incrementalMode: 'delta' | 'full' | null
+  incrementalInputMessages: number
   incrementalMessages: number
+  incrementalDuplicateMessages: number
+  incrementalAnalyzeDurationMs: number
+  incrementalAnalyzeRssDeltaMb: number
   incrementalDurationMs: number
+  incrementalImportRssDeltaMb: number
   sampledPeakRssMb: number
   sampledRssDeltaMb: number
   databaseBytes: number
@@ -204,7 +214,11 @@ async function generateWeflowFixture(filePath: string, count: number, startIndex
   await once(stream, 'finish')
 }
 
-function createIncrementalDeps(dbPath: string): IncrementalImportDeps {
+function createIncrementalDeps(
+  dbPath: string,
+  onSampleRss: () => void,
+  onParserLog: NonNullable<IncrementalImportDeps['onParserLog']>
+): IncrementalImportDeps {
   return {
     openDatabase() {
       const raw = new Database(dbPath, { nativeBinding })
@@ -213,8 +227,9 @@ function createIncrementalDeps(dbPath: string): IncrementalImportDeps {
       return new BetterSqliteAdapter(raw)
     },
     onProgress() {
-      /* benchmark excludes UI transport */
+      onSampleRss()
     },
+    onParserLog,
   }
 }
 
@@ -342,7 +357,12 @@ function measureLikeSearch(dbPath: string): BenchmarkResult['likeSearch'] {
   }
 }
 
-async function runWorker(fileCount: number, messagesPerFile: number, concurrency: number): Promise<BenchmarkResult> {
+async function runWorker(
+  fileCount: number,
+  messagesPerFile: number,
+  concurrency: number,
+  incrementalMode: 'delta' | 'full' | null
+): Promise<BenchmarkResult> {
   if (typeof global.gc !== 'function') {
     throw new Error('Benchmark worker requires Node.js --expose-gc')
   }
@@ -357,8 +377,19 @@ async function runWorker(fileCount: number, messagesPerFile: number, concurrency
   try {
     await generateWeflowFixture(fixturePath, messagesPerFile)
     const incrementalMessages = fileCount === 1 ? 1000 : 0
+    const incrementalInputMessages =
+      incrementalMessages === 0
+        ? 0
+        : incrementalMode === 'full'
+          ? messagesPerFile + incrementalMessages
+          : incrementalMessages
+    const incrementalDuplicateMessages = incrementalMode === 'full' ? messagesPerFile : 0
     if (incrementalMessages > 0) {
-      await generateWeflowFixture(incrementalFixturePath, incrementalMessages, messagesPerFile)
+      if (incrementalMode === 'full') {
+        await generateWeflowFixture(incrementalFixturePath, incrementalInputMessages)
+      } else {
+        await generateWeflowFixture(incrementalFixturePath, incrementalMessages, messagesPerFile)
+      }
     }
     const inputPaths = [fixturePath]
     for (let index = 1; index < fileCount; index++) {
@@ -406,16 +437,62 @@ async function runWorker(fileCount: number, messagesPerFile: number, concurrency
     )
     const durationMs = performance.now() - startedAt
 
+    let incrementalAnalyzeDurationMs = 0
+    let incrementalAnalyzeRssDeltaMb = 0
     let incrementalDurationMs = 0
+    let incrementalImportRssDeltaMb = 0
     if (incrementalMessages > 0) {
+      global.gc()
+      let rssStartMb = process.memoryUsage().rss / 1024 / 1024
+      let rssPeakMb = rssStartMb
+      const sampleIncrementalRss = () => {
+        rssPeakMb = Math.max(rssPeakMb, process.memoryUsage().rss / 1024 / 1024)
+      }
+      const incrementalParserMonitor = createBenchmarkParserMonitor()
+      const incrementalDeps = createIncrementalDeps(dbPaths[0], sampleIncrementalRss, (_level, message) => {
+        incrementalParserMonitor.logger.info(message)
+      })
+
+      incrementalParserMonitor.logger.reset()
+      const analyzeStartedAt = performance.now()
+      const analysis = await analyzeIncrementalImport('bench-0', incrementalFixturePath, incrementalDeps, {
+        formatId: 'weflow',
+      })
+      incrementalAnalyzeDurationMs = performance.now() - analyzeStartedAt
+      incrementalParserMonitor.assertRustNativeCompleted(0)
+      sampleIncrementalRss()
+      incrementalAnalyzeRssDeltaMb = Math.max(0, rssPeakMb - rssStartMb)
+      if (
+        analysis.newMessageCount !== incrementalMessages ||
+        analysis.duplicateCount !== incrementalDuplicateMessages ||
+        analysis.totalInFile !== incrementalInputMessages
+      ) {
+        throw new Error(
+          `Incremental analysis mismatch: expected ${incrementalMessages} new/${incrementalDuplicateMessages} duplicate/${incrementalInputMessages} total, ` +
+            `got ${analysis.newMessageCount}/${analysis.duplicateCount}/${analysis.totalInFile}`
+        )
+      }
+
+      global.gc()
+      rssStartMb = process.memoryUsage().rss / 1024 / 1024
+      rssPeakMb = rssStartMb
+      incrementalParserMonitor.logger.reset()
       const incrementalStartedAt = performance.now()
-      const result = await incrementalImport('bench-0', incrementalFixturePath, createIncrementalDeps(dbPaths[0]), {
+      const result = await incrementalImport('bench-0', incrementalFixturePath, incrementalDeps, {
         formatId: 'weflow',
       })
       incrementalDurationMs = performance.now() - incrementalStartedAt
-      if (!result.success || result.newMessageCount !== incrementalMessages) {
+      incrementalParserMonitor.assertRustNativeCompleted(0)
+      sampleIncrementalRss()
+      incrementalImportRssDeltaMb = Math.max(0, rssPeakMb - rssStartMb)
+      if (
+        !result.success ||
+        result.newMessageCount !== incrementalMessages ||
+        result.batch?.duplicateCount !== incrementalDuplicateMessages
+      ) {
         throw new Error(
-          `Incremental import failed: expected ${incrementalMessages}, got ${result.newMessageCount} (${result.error ?? 'unknown error'})`
+          `Incremental import failed: expected ${incrementalMessages} new/${incrementalDuplicateMessages} duplicate, ` +
+            `got ${result.newMessageCount}/${result.batch?.duplicateCount ?? 'missing'} (${result.error ?? 'unknown error'})`
         )
       }
     }
@@ -442,8 +519,14 @@ async function runWorker(fileCount: number, messagesPerFile: number, concurrency
       inputMessages: fileCount * messagesPerFile,
       inputBytes: statSync(fixturePath).size * fileCount,
       durationMs,
+      incrementalMode,
+      incrementalInputMessages,
       incrementalMessages,
+      incrementalDuplicateMessages,
+      incrementalAnalyzeDurationMs,
+      incrementalAnalyzeRssDeltaMb,
       incrementalDurationMs,
+      incrementalImportRssDeltaMb,
       sampledPeakRssMb,
       sampledRssDeltaMb: Math.max(0, sampledPeakRssMb - rssStartMb),
       databaseBytes,
@@ -458,7 +541,12 @@ async function runWorker(fileCount: number, messagesPerFile: number, concurrency
   }
 }
 
-function runIsolated(fileCount: number, messagesPerFile: number, concurrency: number): BenchmarkResult {
+function runIsolated(
+  fileCount: number,
+  messagesPerFile: number,
+  concurrency: number,
+  incrementalMode: 'delta' | 'full' | null
+): BenchmarkResult {
   const child = spawnSync(
     process.execPath,
     [
@@ -469,6 +557,7 @@ function runIsolated(fileCount: number, messagesPerFile: number, concurrency: nu
       String(fileCount),
       String(messagesPerFile),
       String(concurrency),
+      incrementalMode ?? 'none',
     ],
     {
       cwd: process.cwd(),
@@ -489,10 +578,58 @@ function formatBytes(bytes: number): string {
   return `${(bytes / 1024 / 1024).toFixed(1)} MB`
 }
 
+function medianNumber(values: number[]): number {
+  const sorted = [...values].sort((left, right) => left - right)
+  const midpoint = Math.floor(sorted.length / 2)
+  return sorted.length % 2 === 0 ? (sorted[midpoint - 1] + sorted[midpoint]) / 2 : sorted[midpoint]
+}
+
+function buildMedianResult(results: BenchmarkResult[]): BenchmarkResult {
+  const representative = results[0]
+  const medianTiming = (key: keyof ImportStageTimings) => medianNumber(results.map((result) => result.timings[key]))
+
+  return {
+    ...representative,
+    inputBytes: medianNumber(results.map((result) => result.inputBytes)),
+    durationMs: medianNumber(results.map((result) => result.durationMs)),
+    incrementalAnalyzeDurationMs: medianNumber(results.map((result) => result.incrementalAnalyzeDurationMs)),
+    incrementalAnalyzeRssDeltaMb: medianNumber(results.map((result) => result.incrementalAnalyzeRssDeltaMb)),
+    incrementalDurationMs: medianNumber(results.map((result) => result.incrementalDurationMs)),
+    incrementalImportRssDeltaMb: medianNumber(results.map((result) => result.incrementalImportRssDeltaMb)),
+    sampledPeakRssMb: medianNumber(results.map((result) => result.sampledPeakRssMb)),
+    sampledRssDeltaMb: medianNumber(results.map((result) => result.sampledRssDeltaMb)),
+    databaseBytes: medianNumber(results.map((result) => result.databaseBytes)),
+    likeSearch: {
+      ...representative.likeSearch,
+      coldMs: medianNumber(results.map((result) => result.likeSearch.coldMs)),
+      warmMedianMs: medianNumber(results.map((result) => result.likeSearch.warmMedianMs)),
+    },
+    timings: {
+      detectionMs: medianTiming('detectionMs'),
+      preprocessingMs: medianTiming('preprocessingMs'),
+      databaseSetupMs: medianTiming('databaseSetupMs'),
+      parserMs: medianTiming('parserMs'),
+      metaWriteMs: medianTiming('metaWriteMs'),
+      memberWriteMs: medianTiming('memberWriteMs'),
+      messageWriteMs: medianTiming('messageWriteMs'),
+      nicknameHistoryMs: medianTiming('nicknameHistoryMs'),
+      indexCreationMs: medianTiming('indexCreationMs'),
+      checkpointMs: medianTiming('checkpointMs'),
+      sessionIndexMs: medianTiming('sessionIndexMs'),
+      postImportHookMs: medianTiming('postImportHookMs'),
+      totalMs: medianTiming('totalMs'),
+    },
+  }
+}
+
 function printResult(result: BenchmarkResult, label: string): void {
   console.log(
     `${label}: parser ${result.parser.implementation} | ${result.durationMs.toFixed(0)} ms | ` +
-      `incremental ${result.incrementalDurationMs.toFixed(0)} ms/${result.incrementalMessages} rows | ` +
+      `incremental ${result.incrementalMode ?? 'none'} ` +
+      `${result.incrementalAnalyzeDurationMs.toFixed(0)} ms analyze/${result.incrementalDurationMs.toFixed(0)} ms import ` +
+      `(${result.incrementalMessages} new/${result.incrementalDuplicateMessages} duplicate) | ` +
+      `incremental RSS +${result.incrementalAnalyzeRssDeltaMb.toFixed(0)} MB analyze/` +
+      `+${result.incrementalImportRssDeltaMb.toFixed(0)} MB import | ` +
       `sampled peak RSS ${result.sampledPeakRssMb.toFixed(0)} MB ` +
       `(+${result.sampledRssDeltaMb.toFixed(0)} MB) | DB ${formatBytes(result.databaseBytes)} | ` +
       `LIKE ${result.likeSearch.coldMs.toFixed(1)} ms cold/${result.likeSearch.warmMedianMs.toFixed(1)} ms warm | ` +
@@ -503,7 +640,8 @@ function printResult(result: BenchmarkResult, label: string): void {
 async function main(): Promise<void> {
   const [mode, first, second, third, fourth] = process.argv.slice(2)
   if (mode === '--worker') {
-    const result = await runWorker(Number(first), Number(second), Number(third))
+    const workerIncrementalMode = fourth === 'delta' || fourth === 'full' ? fourth : null
+    const result = await runWorker(Number(first), Number(second), Number(third), workerIncrementalMode)
     console.log(`BENCH_RESULT ${JSON.stringify(result)}`)
     return
   }
@@ -512,18 +650,27 @@ async function main(): Promise<void> {
   let messagesPerFile: number
   let runs: number
   let concurrency: number
+  let incrementalMode: 'delta' | 'full' | null
   if (mode === 'single') {
     fileCount = 1
     messagesPerFile = Number(first ?? 100_000)
     runs = Number(second ?? 3)
     concurrency = 1
+    incrementalMode = third === 'full' ? 'full' : 'delta'
+    if (third !== undefined && third !== 'delta' && third !== 'full') {
+      throw new Error('Incremental mode must be delta or full')
+    }
   } else if (mode === 'batch') {
     fileCount = Number(first ?? 100)
     messagesPerFile = Number(second ?? 10_000)
     runs = Number(third ?? 3)
     concurrency = Number(fourth ?? 1)
+    incrementalMode = null
   } else {
-    throw new Error('Usage: single <messages> [runs=3] | batch <files> <messages-per-file> [runs=3] [concurrency=1]')
+    throw new Error(
+      'Usage: single <messages> [runs=3] [incremental-mode=delta|full] | ' +
+        'batch <files> <messages-per-file> [runs=3] [concurrency=1]'
+    )
   }
   if (![fileCount, messagesPerFile, runs, concurrency].every((value) => Number.isInteger(value) && value > 0)) {
     throw new Error('All benchmark counts must be positive integers')
@@ -534,13 +681,12 @@ async function main(): Promise<void> {
   )
   const results: BenchmarkResult[] = []
   for (let index = 0; index < runs; index++) {
-    const result = runIsolated(fileCount, messagesPerFile, concurrency)
+    const result = runIsolated(fileCount, messagesPerFile, concurrency, incrementalMode)
     results.push(result)
     printResult(result, `run ${index + 1}`)
   }
 
-  const sorted = [...results].sort((left, right) => left.durationMs - right.durationMs)
-  const median = sorted[Math.floor(sorted.length / 2)]
+  const median = buildMedianResult(results)
   const parserSignatures = new Set(results.map((result) => JSON.stringify(result.parser)))
   if (parserSignatures.size !== 1) {
     throw new Error('Parser implementations differ between isolated runs')