fix(import): generate session index once

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

@@ -63,10 +63,12 @@ describe('CLI Web archive import source routes', () => {
 
       const manager = new ArchiveImportSourceManager({ tempRoot: join(dir, 'sources') })
       const importedManifests: string[] = []
+      const receivedThresholds: Array<number | undefined> = []
       registerImportRoutes(app, {} as any, {
         sourceManager: manager,
-        runPreparedImport: async (manifestPath) => {
+        runPreparedImport: async (manifestPath, _onProgress, sessionGapThreshold) => {
           importedManifests.push(readFileSync(manifestPath, 'utf8'))
+          receivedThresholds.push(sessionGapThreshold)
           return { success: true, sessionId: `session-${importedManifests.length}` }
         },
       })
@@ -88,13 +90,14 @@ describe('CLI Web archive import source routes', () => {
         const response = await app.inject({
           method: 'POST',
           url: `/_web/import-sources/${prepared.source.sourceId}/import`,
-          payload: { chatId: 'Groups/DM sample' },
+          payload: { chatId: 'Groups/DM sample', sessionGapThreshold: 7200 },
         })
         assert.equal(response.statusCode, 200)
         assert.match(response.body, /event: done/)
         assert.match(response.body, new RegExp(`session-${index + 1}`))
       }
       assert.equal(importedManifests.length, 2)
+      assert.deepEqual(receivedThresholds, [7200, 7200])
 
       for (let index = 0; index < 2; index++) {
         const response = await app.inject({

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

@@ -19,7 +19,14 @@ function multipartPayload(): { payload: Buffer; contentType: string } {
 
   return {
     payload: Buffer.from(
-      [field('formatId', 'telegram-json'), field('chatIndex', '2'), file, `--${boundary}--`, ''].join('\r\n')
+      [
+        field('formatId', 'telegram-json'),
+        field('chatIndex', '2'),
+        field('sessionGapThreshold', '7200'),
+        file,
+        `--${boundary}--`,
+        '',
+      ].join('\r\n')
     ),
     contentType: `multipart/form-data; boundary=${boundary}`,
   }
@@ -56,7 +63,7 @@ function directoryMultipartPayload(): { payload: Buffer; contentType: string } {
 describe('CLI Web automatic import route', () => {
   it('forwards parser options and preserves a zero-new incremental result in the done event', async () => {
     const app = Fastify()
-    const calls: Array<{ formatId?: string; chatIndex?: number }> = []
+    const calls: Array<{ formatId?: string; chatIndex?: number; sessionGapThreshold?: number }> = []
     try {
       await app.register(multipart)
       registerImportRoutes(
@@ -67,6 +74,7 @@ describe('CLI Web automatic import route', () => {
             calls.push({
               formatId: options.formatId as string | undefined,
               chatIndex: options.chatIndex as number | undefined,
+              sessionGapThreshold: options.sessionGapThreshold as number | undefined,
             })
             return {
               success: true,
@@ -89,7 +97,7 @@ describe('CLI Web automatic import route', () => {
       })
 
       assert.equal(response.statusCode, 200)
-      assert.deepEqual(calls, [{ formatId: 'telegram-json', chatIndex: 2 }])
+      assert.deepEqual(calls, [{ formatId: 'telegram-json', chatIndex: 2, sessionGapThreshold: 7200 }])
       assert.match(response.body, /event: done/)
       assert.doesNotMatch(response.body, /event: error/)
       assert.match(response.body, /"importMode":"incremental"/)

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

@@ -33,10 +33,18 @@ interface ImportRouteOptions {
   runAutoImport?: (filePath: string, options: StreamImportOptions) => Promise<AutoImportResult>
   runPreparedImport?: (
     manifestPath: string,
-    onProgress: (progress: unknown) => void
+    onProgress: (progress: unknown) => void,
+    sessionGapThreshold?: number
   ) => Promise<{ success: boolean; sessionId?: string; error?: string; messageCount?: number; memberCount?: number }>
 }
 
+function parseOptionalInteger(value: unknown): number | undefined {
+  if (typeof value === 'number') return Number.isInteger(value) ? value : undefined
+  if (typeof value !== 'string' || value.trim() === '') return undefined
+  const parsed = Number(value)
+  return Number.isInteger(parsed) ? parsed : undefined
+}
+
 function cleanupTemp(...paths: string[]) {
   for (const p of paths) {
     try {
@@ -61,9 +69,10 @@ export function registerImportRoutes(
   const runAutoImport = options.runAutoImport ?? autoImport.bind(null, dbManager)
   const runPreparedImport =
     options.runPreparedImport ??
-    (async (manifestPath: string, onProgress: (progress: unknown) => void) => {
+    (async (manifestPath: string, onProgress: (progress: unknown) => void, sessionGapThreshold?: number) => {
       const result = await runAutoImport(manifestPath, {
         formatId: 'google-chat-takeout',
+        sessionGapThreshold,
         nativeBinding: resolveNativeBinding(),
         onProgress: onProgress as any,
       })
@@ -102,7 +111,7 @@ export function registerImportRoutes(
     }
   })
 
-  server.post<{ Params: { sourceId: string }; Body: { chatId?: string } }>(
+  server.post<{ Params: { sourceId: string }; Body: { chatId?: string; sessionGapThreshold?: number } }>(
     '/_web/import-sources/:sourceId/import',
     async (request, reply) => {
       const chatId = request.body?.chatId
@@ -119,8 +128,9 @@ export function registerImportRoutes(
       }
 
       try {
+        const sessionGapThreshold = parseOptionalInteger(request.body?.sessionGapThreshold)
         const result = await sourceManager.withMaterializedChat(request.params.sourceId, chatId, (manifestPath) =>
-          runPreparedImport(manifestPath, (progress) => sendEvent('progress', progress))
+          runPreparedImport(manifestPath, (progress) => sendEvent('progress', progress), sessionGapThreshold)
         )
         if (result.success) {
           sendEvent('done', result)
@@ -207,6 +217,7 @@ export function registerImportRoutes(
     const formatId = (data.fields?.formatId as any)?.value as string | undefined
     const chatIndexStr = (data.fields?.chatIndex as any)?.value as string | undefined
     const chatIndex = chatIndexStr !== undefined ? parseInt(chatIndexStr, 10) : undefined
+    const sessionGapThreshold = parseOptionalInteger((data.fields?.sessionGapThreshold as any)?.value)
 
     reply.raw.writeHead(200, {
       'Content-Type': 'text/event-stream',
@@ -223,6 +234,7 @@ export function registerImportRoutes(
       const result = await runAutoImport(tmpPath, {
         formatId,
         chatIndex,
+        sessionGapThreshold,
         nativeBinding,
         onProgress: (p) => sendEvent('progress', p),
       })
@@ -249,11 +261,14 @@ export function registerImportRoutes(
     const tmpDir = createChatLabTempDir('imports', 'directory-')
     const relativePaths: string[] = []
     const fileBuffers: { data: Buffer; filename: string }[] = []
+    let sessionGapThreshold: number | undefined
 
     try {
       for await (const part of parts) {
         if (part.type === 'field' && part.fieldname === 'relativePaths') {
           relativePaths.push(String(part.value))
+        } else if (part.type === 'field' && part.fieldname === 'sessionGapThreshold') {
+          sessionGapThreshold = parseOptionalInteger(part.value)
         } else if (part.type === 'file') {
           const chunks: Buffer[] = []
           for await (const chunk of part.file) {
@@ -292,6 +307,7 @@ export function registerImportRoutes(
 
       const nativeBinding = resolveNativeBinding()
       const result = await runAutoImport(entryPath, {
+        sessionGapThreshold,
         nativeBinding,
         onProgress: (p) => sendEvent('progress', p),
       })
@@ -407,32 +423,36 @@ export function registerImportRoutes(
 
   // ==================== Demo Import ====================
 
-  server.post<{ Body: { locale?: string; timeZone?: string } }>('/_web/demo/import', async (request, reply) => {
-    const locale = request.body?.locale === 'cn' ? 'cn' : 'en'
-    const targetTimeZone = typeof request.body?.timeZone === 'string' ? request.body.timeZone : undefined
-    const nativeBinding = resolveNativeBinding()
+  server.post<{ Body: { locale?: string; timeZone?: string; sessionGapThreshold?: number } }>(
+    '/_web/demo/import',
+    async (request, reply) => {
+      const locale = request.body?.locale === 'cn' ? 'cn' : 'en'
+      const targetTimeZone = typeof request.body?.timeZone === 'string' ? request.body.timeZone : undefined
+      const sessionGapThreshold = parseOptionalInteger(request.body?.sessionGapThreshold)
+      const nativeBinding = resolveNativeBinding()
 
-    reply.raw.writeHead(200, {
-      'Content-Type': 'text/event-stream',
-      'Cache-Control': 'no-cache',
-      Connection: 'keep-alive',
-    })
+      reply.raw.writeHead(200, {
+        'Content-Type': 'text/event-stream',
+        'Cache-Control': 'no-cache',
+        Connection: 'keep-alive',
+      })
 
-    function sendEvent(event: string, eventData: unknown) {
-      reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(eventData)}\n\n`)
+      function sendEvent(event: string, eventData: unknown) {
+        reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(eventData)}\n\n`)
+      }
+
+      const result = await importDemoSessions({
+        locale,
+        tempPrefix: 'cli-demo-',
+        targetTimeZone,
+        importFile: (filePath) => streamImport(dbManager, filePath, { nativeBinding, sessionGapThreshold }),
+        deleteSession: (sessionId) => {
+          dbManager.deleteSessionDatabaseFiles(sessionId)
+        },
+        onProgress: (progress) => sendEvent('progress', progress),
+      })
+      sendEvent('result', result)
+      reply.raw.end()
     }
-
-    const result = await importDemoSessions({
-      locale,
-      tempPrefix: 'cli-demo-',
-      targetTimeZone,
-      importFile: (filePath) => streamImport(dbManager, filePath, { nativeBinding }),
-      deleteSession: (sessionId) => {
-        dbManager.deleteSessionDatabaseFiles(sessionId)
-      },
-      onProgress: (progress) => sendEvent('progress', progress),
-    })
-    sendEvent('result', result)
-    reply.raw.end()
-  })
+  )
 }

apps/cli/src/import/command.ts

@@ -174,6 +174,7 @@ export function registerImportCommand(program: Command): void {
           const importOptions = {
             formatId: options.format,
             sessionId: options.sessionId,
+            sessionGapThreshold: runtime.config.ui.session_gap_threshold,
             nativeBinding,
             onProgress: options.json
               ? undefined

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

@@ -76,13 +76,27 @@ test('streamImport raises the data directory gate after creating a current-schem
   })
 
   const chatFile = writeTempChatFile(root)
-  const result = await streamImport(manager, chatFile, { sessionId: 'test-session', nativeBinding })
+  const result = await streamImport(manager, chatFile, {
+    sessionId: 'test-session',
+    sessionGapThreshold: 7200,
+    nativeBinding,
+  })
 
   assert.equal(result.success, true)
   const meta = readDataDirCompatibilityMeta(path.join(root, 'data'))
   assert.equal(meta?.minRuntimeVersion, '0.25.1')
   assert.equal(meta?.dataCompatibilityVersion, 1)
   assert.deepEqual(meta?.reasons, ['segment-schema'])
+
+  const db = new Database(path.join(root, 'data', 'databases', 'test-session.db'), {
+    readonly: true,
+    nativeBinding,
+  })
+  const sessionMeta = db.prepare('SELECT session_gap_threshold FROM meta LIMIT 1').get() as {
+    session_gap_threshold: number | null
+  }
+  db.close()
+  assert.equal(sessionMeta.session_gap_threshold, 7200)
 })
 
 test('streamImport re-checks data directory compatibility before raw database writes', async () => {

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

@@ -58,6 +58,7 @@ export interface StreamImportProgress {
 export interface StreamImportOptions {
   formatId?: string
   chatIndex?: number
+  sessionGapThreshold?: number
   nativeBinding?: string
   onProgress?: (progress: StreamImportProgress) => void
   /** Fix the target session ID instead of auto-generating one. Used by sync/pull adapters. */
@@ -70,7 +71,11 @@ function generateSessionId(): string {
   return `chat_${ts}_${rand}`
 }
 
-function buildStreamImportDeps(dbManager: DatabaseManager, onProgress?: ImportProgressCallback): StreamImportDeps {
+function buildStreamImportDeps(
+  dbManager: DatabaseManager,
+  onProgress?: ImportProgressCallback,
+  sessionGapThreshold?: number
+): StreamImportDeps {
   return {
     openDatabase(sessionId: string) {
       return dbManager.openRawSessionDatabase(sessionId, { create: true, initializeChatTables: true })
@@ -80,6 +85,7 @@ function buildStreamImportDeps(dbManager: DatabaseManager, onProgress?: ImportPr
     },
     onProgress: onProgress ?? (() => {}),
     generateSessionId,
+    sessionGapThreshold,
   }
 }
 
@@ -138,7 +144,7 @@ async function streamImportUnlocked(
   filePath: string,
   options?: StreamImportOptions
 ): Promise<StreamImportResult> {
-  const { formatId, chatIndex, onProgress, sessionId } = options || {}
+  const { formatId, chatIndex, sessionGapThreshold, onProgress, sessionId } = options || {}
 
   const formatOptions: Record<string, unknown> = {}
   if (formatId) formatOptions.formatId = formatId
@@ -146,7 +152,7 @@ async function streamImportUnlocked(
 
   const progressAdapter = createProgressAdapter(onProgress)
 
-  const deps = buildStreamImportDeps(dbManager, progressAdapter)
+  const deps = buildStreamImportDeps(dbManager, progressAdapter, sessionGapThreshold)
   const result = await streamingImport(filePath, deps, formatOptions, sessionId)
   if (!result.success || !result.sessionId) return result