refactor(import): route desktop imports through shared services

apps/desktop/main/api/errors.ts

@@ -71,6 +71,13 @@ export function idempotencyConflict(): ApiError {
   return new ApiError(ApiErrorCode.IDEMPOTENCY_CONFLICT, 'Same Idempotency-Key with different request body hash')
 }
 
+export function idempotencyPending(): ApiError {
+  return new ApiError(
+    ApiErrorCode.IDEMPOTENCY_PENDING,
+    'A request with this Idempotency-Key is still in progress. Please retry later.'
+  )
+}
+
 export function sqlReadonlyViolation(): ApiError {
   return new ApiError(ApiErrorCode.SQL_READONLY_VIOLATION, 'Only SELECT queries are allowed')
 }

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, batchFromStreamDiagnostics } from './import-helpers'
+import { analysisFromNewImport, 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')
@@ -19,30 +19,6 @@ test('keeps ordinary import failures as IMPORT_FAILED', () => {
   assert.equal(error.message, 'database write failed')
 })
 
-test('maps stream duplicate and invalid-row counts to their distinct API fields', () => {
-  assert.deepEqual(
-    batchFromStreamDiagnostics({
-      logFile: null,
-      detectedFormat: 'fixture',
-      messagesReceived: 12,
-      messagesWritten: 7,
-      duplicateCount: 3,
-      messagesSkipped: 2,
-      skipReasons: {
-        noSenderId: 1,
-        noAccountName: 0,
-        invalidTimestamp: 1,
-        noType: 0,
-      },
-    }),
-    {
-      receivedCount: 12,
-      writtenCount: 7,
-      duplicateCount: 3,
-    }
-  )
-})
-
 test('maps new-import analysis counts without assuming every parsed message is new', () => {
   assert.deepEqual(
     analysisFromNewImport({

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

@@ -1,26 +1,11 @@
 import { IMPORT_IN_PROGRESS_ERROR_KEY } from '@openchatlab/node-runtime/src/import/import-lock'
-import type { AnalyzeNewImportResult, ImportDiagnostics } from '@openchatlab/node-runtime/src/import/streaming-importer'
+import type { AnalyzeNewImportResult } from '@openchatlab/node-runtime/src/import/streaming-importer'
 import { importFailed, importInProgress, type ApiError } from '../errors'
 
 export function apiErrorFromImportResult(error: string | undefined, fallbackMessage: string): ApiError {
   return error === IMPORT_IN_PROGRESS_ERROR_KEY ? importInProgress() : importFailed(error || fallbackMessage)
 }
 
-export function batchFromStreamDiagnostics(diagnostics: ImportDiagnostics | undefined):
-  | {
-      receivedCount: number
-      writtenCount: number
-      duplicateCount: number
-    }
-  | undefined {
-  if (!diagnostics) return undefined
-  return {
-    receivedCount: diagnostics.messagesReceived,
-    writtenCount: diagnostics.messagesWritten,
-    duplicateCount: diagnostics.duplicateCount,
-  }
-}
-
 export function analysisFromNewImport(result: AnalyzeNewImportResult): {
   totalInFile: number
   newMessageCount: number

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

@@ -8,8 +8,8 @@
  * POST /api/v1/sessions/:id/import Incremental import to existing session
  *
  * Content-Type dispatch:
- *   application/json     → parse body → temp .json → chatlab parser
- *   application/x-ndjson → pipe raw stream → temp .jsonl → chatlab-jsonl parser
+ *   application/json     → shared pushImport service in the Desktop worker
+ *   application/x-ndjson → raw stream → temp .jsonl → shared auto-import orchestration
  */
 
 import type { FastifyInstance, FastifyRequest, FastifyReply } from 'fastify'
@@ -18,6 +18,8 @@ import * as fs from 'fs'
 import * as path from 'path'
 import * as crypto from 'crypto'
 import { pipeline } from 'stream/promises'
+import { hashImportBody, ImportIdempotencyCache, isValidImportSessionId } from '@openchatlab/node-runtime'
+import type { PushImportPayload } from '@openchatlab/node-runtime'
 import { getTempDir } from '../../paths'
 import * as worker from '../../worker/workerManager'
 import {
@@ -30,45 +32,22 @@ import {
   invalidFormat,
   invalidPayload,
   idempotencyConflict,
+  idempotencyPending,
   errorResponse,
 } from '../errors'
 import { apiLogger } from '../logger'
-import { analysisFromNewImport, apiErrorFromImportResult, batchFromStreamDiagnostics } from './import-helpers'
+import { analysisFromNewImport, apiErrorFromImportResult } from './import-helpers'
 
-// Per-session lock: different sessionIds can import in parallel.
+// Tracks active External API requests; the shared data-directory lock enforces writer exclusion.
 const isImporting = new Set<string>()
 
-// ==================== Idempotency cache ====================
+type ImportSuccessResponse = ReturnType<typeof successResponse>
 
-interface IdempotencyCacheEntry {
-  bodyHash: string
-  status: 'pending' | 'success'
-  response: any
-  timestamp: number
-}
+const idempotencyCache = new ImportIdempotencyCache<ImportSuccessResponse>()
 
-const IDEMPOTENCY_TTL_MS = 60 * 60 * 1000 // 1 hour
-const IDEMPOTENCY_CLEANUP_INTERVAL_MS = 10 * 60 * 1000 // 10 minutes
-const idempotencyCache = new Map<string, IdempotencyCacheEntry>()
-
-setInterval(() => {
-  const now = Date.now()
-  for (const [key, entry] of idempotencyCache) {
-    if (now - entry.timestamp > IDEMPOTENCY_TTL_MS) {
-      idempotencyCache.delete(key)
-    }
-  }
-}, IDEMPOTENCY_CLEANUP_INTERVAL_MS)
-
-function computeBodyHash(body: unknown, tempFile?: string): string {
-  if (tempFile && fs.existsSync(tempFile)) {
-    const content = fs.readFileSync(tempFile)
-    return crypto.createHash('sha256').update(content).digest('hex')
-  }
-  return crypto
-    .createHash('sha256')
-    .update(JSON.stringify(body ?? ''))
-    .digest('hex')
+function computeTempFileHash(tempFile: string): string {
+  const content = fs.readFileSync(tempFile)
+  return crypto.createHash('sha256').update(content).digest('hex')
 }
 
 function getTempFilePath(ext: string): string {
@@ -97,18 +76,14 @@ function notifySessionListChanged(): void {
   }
 }
 
-function idempotencySuccess(key: string | undefined, response: any): void {
+function idempotencySuccess(key: string | undefined, response: ImportSuccessResponse): void {
   if (!key) return
-  const entry = idempotencyCache.get(key)
-  if (entry) {
-    entry.status = 'success'
-    entry.response = response
-  }
+  idempotencyCache.success(key, response)
 }
 
 function idempotencyFail(key: string | undefined): void {
   if (!key) return
-  idempotencyCache.delete(key)
+  idempotencyCache.fail(key)
 }
 
 export function getImportingStatus(sessionId?: string): boolean {
@@ -157,12 +132,6 @@ async function writeTempFile(
  * v3 统一导入处理：自动判断新建或增量
  */
 async function handleUnifiedImport(request: FastifyRequest, reply: FastifyReply, sessionId: string): Promise<void> {
-  if (isImporting.has(sessionId)) {
-    const err = importInProgress()
-    reply.code(err.statusCode).send(errorResponse(err))
-    return
-  }
-
   const contentType = (request.headers['content-type'] || '').toLowerCase()
   const isJsonl = contentType.includes('application/x-ndjson')
   const isJson = contentType.includes('application/json')
@@ -178,50 +147,58 @@ async function handleUnifiedImport(request: FastifyRequest, reply: FastifyReply,
 
   const cacheKey = idempotencyKey ? `${idempotencyKey}:${sessionId}:${isDryRun}` : undefined
 
-  isImporting.add(sessionId)
   let tempFile = ''
+  let activeRequestRegistered = false
+  let idempotencyOwned = false
 
   try {
-    const writeResult = await writeTempFile(request, isJson)
-    if (writeResult.error) {
-      const err = invalidFormat(writeResult.error)
+    // 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) {
+      const writeResult = await writeTempFile(request, isJson)
+      if (writeResult.error) {
+        const err = invalidFormat(writeResult.error)
+        reply.code(err.statusCode).send(errorResponse(err))
+        return
+      }
+      tempFile = writeResult.tempFile!
+    }
+
+    if (cacheKey) {
+      const bodyHash = isJsonl ? computeTempFileHash(tempFile) : hashImportBody(request.body)
+      const start = idempotencyCache.start(cacheKey, bodyHash)
+      if (start.status === 'conflict') {
+        const err = idempotencyConflict()
+        reply.code(err.statusCode).send(errorResponse(err))
+        return
+      }
+      if (start.status === 'pending') {
+        const err = idempotencyPending()
+        reply.code(err.statusCode).send(errorResponse(err))
+        return
+      }
+      if (start.status === 'success') {
+        reply.send(start.response)
+        return
+      }
+      idempotencyOwned = true
+    }
+
+    // 幂等状态必须先于本地活跃请求检查，确保同 key 并发重试返回
+    // IDEMPOTENCY_PENDING；只有本次新占位的 key 才会在锁冲突时释放。
+    if (isImporting.has(sessionId)) {
+      idempotencyFail(cacheKey)
+      const err = importInProgress()
       reply.code(err.statusCode).send(errorResponse(err))
       return
     }
-    tempFile = writeResult.tempFile!
-
-    // Idempotency-Key check (after tempFile is written so we can hash JSONL content)
-    if (cacheKey) {
-      const bodyHash = isJsonl ? computeBodyHash(null, tempFile) : computeBodyHash(request.body)
-      const cached = idempotencyCache.get(cacheKey)
-      if (cached) {
-        if (cached.bodyHash !== bodyHash) {
-          const err = idempotencyConflict()
-          reply.code(err.statusCode).send(errorResponse(err))
-          return
-        }
-        if (cached.status === 'pending') {
-          const pendingErr = new ApiError(
-            ApiErrorCode.IDEMPOTENCY_PENDING,
-            'A request with this Idempotency-Key is still in progress. Please retry later.'
-          )
-          reply.code(pendingErr.statusCode).send(errorResponse(pendingErr))
-          return
-        }
-        reply.send(cached.response)
-        return
-      }
-      idempotencyCache.set(cacheKey, { bodyHash, status: 'pending', response: null, timestamp: Date.now() })
-    }
-
-    const importOptions =
-      isJson && request.body && typeof request.body === 'object' ? (request.body as any).options : undefined
-
-    const exists = sessionExists(sessionId)
+    isImporting.add(sessionId)
+    activeRequestRegistered = true
 
     // X-Dry-Run: analyze only, no writes
     if (isDryRun) {
-      let responsePayload: any
+      const exists = sessionExists(sessionId)
+      let responsePayload: ImportSuccessResponse
       if (exists) {
         const result = await worker.analyzeIncrementalImport(sessionId, tempFile)
         if (result.error) {
@@ -256,67 +233,70 @@ async function handleUnifiedImport(request: FastifyRequest, reply: FastifyReply,
         })
       }
       idempotencySuccess(cacheKey, responsePayload)
+      idempotencyOwned = false
       reply.send(responsePayload)
       return
     }
 
-    if (exists) {
-      const result = await worker.incrementalImport(sessionId, tempFile, undefined, importOptions)
-
-      if (result.success) {
-        notifySessionListChanged()
-        const responsePayload = successResponse({
-          sessionId,
-          created: false,
-          batch: result.batch,
-          session: result.session,
-          updates: result.updates,
-        })
-        idempotencySuccess(cacheKey, responsePayload)
-        reply.send(responsePayload)
-      } else {
+    if (isJson) {
+      const outcome = await worker.pushImport(sessionId, (request.body ?? {}) as PushImportPayload)
+      if (!outcome.ok) {
         idempotencyFail(cacheKey)
-        const err = apiErrorFromImportResult(result.error, 'Incremental import failed')
+        const err =
+          outcome.reason === 'import_in_progress'
+            ? importInProgress()
+            : outcome.reason === 'invalid_payload'
+              ? invalidPayload(outcome.message)
+              : importFailed(outcome.message)
         reply.code(err.statusCode).send(errorResponse(err))
+        return
       }
-    } else {
-      const result = await worker.streamImport(tempFile, undefined, undefined, sessionId)
+      notifySessionListChanged()
+      const responsePayload = successResponse(outcome.result)
+      idempotencySuccess(cacheKey, responsePayload)
+      idempotencyOwned = false
+      reply.send(responsePayload)
+      return
+    }
 
-      if (result.success) {
-        notifySessionListChanged()
+    const result = await worker.autoImport(tempFile, undefined, undefined, sessionId)
+    if (!result.success || !result.sessionId || !result.importMode) {
+      idempotencyFail(cacheKey)
+      const err = apiErrorFromImportResult(result.error, 'Import failed')
+      reply.code(err.statusCode).send(errorResponse(err))
+      return
+    }
 
-        const diag = result.diagnostics
-        let sessionInfo: any = undefined
-        try {
-          const s = await worker.getSession(result.sessionId!)
-          if (s) {
-            sessionInfo = {
-              totalCount: s.totalCount,
-              memberCount: s.memberCount,
-              firstTimestamp: s.firstTimestamp,
-              lastTimestamp: s.lastTimestamp,
-            }
+    let sessionInfo = result.session
+    if (!sessionInfo) {
+      try {
+        const session = await worker.getSession(result.sessionId)
+        if (session) {
+          sessionInfo = {
+            totalCount: session.totalCount,
+            memberCount: session.memberCount,
+            firstTimestamp: session.firstTimestamp,
+            lastTimestamp: session.lastTimestamp,
           }
-        } catch {
-          // non-blocking
         }
-
-        const responsePayload = successResponse({
-          sessionId: result.sessionId,
-          created: true,
-          batch: batchFromStreamDiagnostics(diag),
-          session: sessionInfo,
-        })
-        idempotencySuccess(cacheKey, responsePayload)
-        reply.send(responsePayload)
-      } else {
-        idempotencyFail(cacheKey)
-        const err = apiErrorFromImportResult(result.error, 'Import failed')
-        reply.code(err.statusCode).send(errorResponse(err))
+      } catch {
+        // Session statistics are informative and must not turn a successful import into a failure.
       }
     }
+
+    notifySessionListChanged()
+    const responsePayload = successResponse({
+      sessionId: result.sessionId,
+      created: result.importMode === 'created',
+      batch: result.batch,
+      session: sessionInfo,
+      updates: result.updates,
+    })
+    idempotencySuccess(cacheKey, responsePayload)
+    idempotencyOwned = false
+    reply.send(responsePayload)
   } catch (error: any) {
-    idempotencyFail(cacheKey)
+    if (idempotencyOwned) idempotencyFail(cacheKey)
     apiLogger.error('Import error', error)
     const apiError = apiErrorFromUnknown(error)
     if (apiError) {
@@ -326,7 +306,7 @@ async function handleUnifiedImport(request: FastifyRequest, reply: FastifyReply,
     const err = importFailed(error.message || 'Import process error')
     reply.code(err.statusCode).send(errorResponse(err))
   } finally {
-    isImporting.delete(sessionId)
+    if (activeRequestRegistered) isImporting.delete(sessionId)
     if (tempFile) {
       cleanupTempFile(tempFile)
     }
@@ -433,13 +413,13 @@ export function registerImportRoutes(server: FastifyInstance): void {
     done(null, undefined)
   })
 
-  const SESSION_ID_RE = /^[A-Za-z0-9._@-]{1,128}$/
-
   // v3 unified endpoint
   server.post<{ Params: { sessionId: string } }>('/api/v1/imports/:sessionId', async (request, reply) => {
     const { sessionId } = request.params
-    if (!SESSION_ID_RE.test(sessionId)) {
-      const err = invalidPayload('sessionId must be 1-128 characters of [A-Za-z0-9._@-]')
+    if (!isValidImportSessionId(sessionId)) {
+      const err = invalidPayload(
+        "sessionId must be 1-128 safe characters, start with [A-Za-z0-9_@-], and not contain '..'"
+      )
       reply.code(err.statusCode).send(errorResponse(err))
       return
     }

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

@@ -59,6 +59,17 @@ function createDeps(options?: {
           errorReasonCounts: {},
           errorSample: [],
         },
+        session: {
+          totalCount: 12,
+          memberCount: 2,
+          firstTimestamp: 100,
+          lastTimestamp: 200,
+        },
+        updates: {
+          metaUpdated: true,
+          membersAdded: 1,
+          membersUpdated: 1,
+        },
       }
     },
   }
@@ -77,6 +88,25 @@ test('explicit existing session forces incremental import without matching', asy
     importMode: 'incremental',
     newMessageCount: 3,
     duplicateCount: 5,
+    batch: {
+      receivedCount: 8,
+      writtenCount: 3,
+      duplicateCount: 5,
+      errorCount: 0,
+      errorReasonCounts: {},
+      errorSample: [],
+    },
+    session: {
+      totalCount: 12,
+      memberCount: 2,
+      firstTimestamp: 100,
+      lastTimestamp: 200,
+    },
+    updates: {
+      metaUpdated: true,
+      membersAdded: 1,
+      membersUpdated: 1,
+    },
   })
   assert.equal(calls.match, 0)
   assert.deepEqual(calls.create, [])
@@ -93,10 +123,27 @@ test('explicit missing session creates with the requested id without matching',
   assert.equal(result.importMode, 'created')
   assert.equal(result.newMessageCount, 10)
   assert.equal(result.duplicateCount, 0)
+  assert.deepEqual(result.batch, {
+    receivedCount: 12,
+    writtenCount: 10,
+    duplicateCount: 0,
+  })
   assert.equal(calls.match, 0)
   assert.deepEqual(calls.create, [{ filePath: 'source.json', sessionId: 'requested' }])
 })
 
+test('explicit session IDs reject path-like or overlong values before opening a database', async () => {
+  const { deps, calls } = createDeps()
+
+  const traversal = await autoImportFile('source.json', deps, { explicitSessionId: '../outside' })
+  const overlong = await autoImportFile('source.json', deps, { explicitSessionId: 'a'.repeat(129) })
+
+  assert.deepEqual(traversal, { success: false, error: 'sessionId contains invalid characters' })
+  assert.deepEqual(overlong, { success: false, error: 'sessionId contains invalid characters' })
+  assert.deepEqual(calls.create, [])
+  assert.deepEqual(calls.append, [])
+})
+
 test('automatic unique match preserves incremental mode when every message is duplicate', async () => {
   const { deps, calls } = createDeps({
     decision: { action: 'incremental', sessionId: 'existing', matchedBy: 'trailing-messages' },
@@ -113,6 +160,25 @@ test('automatic unique match preserves incremental mode when every message is du
     matchedBy: 'trailing-messages',
     newMessageCount: 0,
     duplicateCount: 500,
+    batch: {
+      receivedCount: 500,
+      writtenCount: 0,
+      duplicateCount: 500,
+      errorCount: 0,
+      errorReasonCounts: {},
+      errorSample: [],
+    },
+    session: {
+      totalCount: 12,
+      memberCount: 2,
+      firstTimestamp: 100,
+      lastTimestamp: 200,
+    },
+    updates: {
+      metaUpdated: true,
+      membersAdded: 1,
+      membersUpdated: 1,
+    },
   })
   assert.equal(calls.match, 1)
   assert.deepEqual(calls.create, [])

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

@@ -8,6 +8,7 @@ import {
 } from './auto-import-matcher'
 import type { IncrementalImportResult } from './incremental-importer'
 import type { ImportDiagnostics, StreamImportResult } from './streaming-importer'
+import { isValidImportSessionId } from './session-id'
 
 export interface AutoImportOptions {
   explicitSessionId?: string
@@ -37,6 +38,11 @@ export interface AutoImportResult {
   createReason?: AutoImportCreateReason
   newMessageCount?: number
   duplicateCount?: number
+  batch?:
+    | NonNullable<IncrementalImportResult['batch']>
+    | { receivedCount: number; writtenCount: number; duplicateCount: number }
+  session?: IncrementalImportResult['session']
+  updates?: IncrementalImportResult['updates']
   diagnostics?: ImportDiagnostics
   error?: string
 }
@@ -52,6 +58,13 @@ function mapCreateResult(result: StreamImportResult, createReason?: AutoImportCr
     ...(createReason ? { createReason } : {}),
     newMessageCount: result.diagnostics?.messagesWritten ?? 0,
     duplicateCount: result.diagnostics?.duplicateCount ?? 0,
+    batch: result.diagnostics
+      ? {
+          receivedCount: result.diagnostics.messagesReceived,
+          writtenCount: result.diagnostics.messagesWritten,
+          duplicateCount: result.diagnostics.duplicateCount,
+        }
+      : undefined,
     diagnostics: result.diagnostics,
   }
 }
@@ -69,6 +82,9 @@ function mapIncrementalResult(
     ...(matchedBy ? { matchedBy } : {}),
     newMessageCount: result.newMessageCount,
     duplicateCount: result.batch?.duplicateCount ?? 0,
+    batch: result.batch,
+    session: result.session,
+    updates: result.updates,
   }
 }
 
@@ -79,6 +95,9 @@ export async function autoImportFile(
 ): Promise<AutoImportResult> {
   try {
     if (options.explicitSessionId) {
+      if (!isValidImportSessionId(options.explicitSessionId)) {
+        return { success: false, error: 'sessionId contains invalid characters' }
+      }
       if (deps.sessionExists(options.explicitSessionId)) {
         const result = mapIncrementalResult(
           options.explicitSessionId,

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

@@ -32,6 +32,8 @@ export type {
   IncrementalImportResult,
   IncrementalImportDeps,
 } from './incremental-importer'
+
+export { isValidImportSessionId } from './session-id'
 export { resolveAutoImportTarget } from './auto-import-matcher'
 export type {
   AutoImportCreateReason,

packages/node-runtime/src/index.ts

@@ -17,6 +17,7 @@ export {
   incrementalImport,
   resolveAutoImportTarget,
   autoImportFile,
+  isValidImportSessionId,
   IMPORT_IN_PROGRESS_ERROR_KEY,
   IMPORT_LOCK_FILENAME,
   ImportInProgressError,
@@ -427,6 +428,10 @@ export {
   createPeopleRelationshipsService,
   createGlobalInsightService,
   createDatabaseManagerAdapter,
+  executePushImportUnlocked,
+  DEFAULT_IMPORT_IDEMPOTENCY_TTL_MS,
+  hashImportBody,
+  ImportIdempotencyCache,
   pushImport,
 } from './services'
 export type {
@@ -448,6 +453,8 @@ export type {
   AnnualSummaryComputeRunner,
   GlobalInsightService,
   GlobalInsightServiceDeps,
+  ImportIdempotencyStartResult,
+  PushImportExecutionDeps,
   GlobalInsightServiceOptions,
   PushImportPayload,
   PushImportResult,

packages/node-runtime/src/services/import-idempotency.test.ts

@@ -0,0 +1,38 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+import { hashImportBody, ImportIdempotencyCache } from './import-idempotency'
+
+test('replays a successful response for the same key and body hash', () => {
+  const cache = new ImportIdempotencyCache<{ ok: boolean }>()
+  const hash = hashImportBody({ messages: [1] })
+
+  assert.deepEqual(cache.start('session:key', hash, 100), { status: 'started' })
+  assert.deepEqual(cache.start('session:key', hash, 101), { status: 'pending' })
+
+  cache.success('session:key', { ok: true })
+  assert.deepEqual(cache.start('session:key', hash, 102), {
+    status: 'success',
+    response: { ok: true },
+  })
+})
+
+test('reports conflicts and releases failed requests for retry', () => {
+  const cache = new ImportIdempotencyCache<string>()
+  const firstHash = hashImportBody({ messages: ['first'] })
+  const secondHash = hashImportBody({ messages: ['second'] })
+
+  assert.deepEqual(cache.start('session:key', firstHash, 100), { status: 'started' })
+  assert.deepEqual(cache.start('session:key', secondHash, 101), { status: 'conflict' })
+
+  cache.fail('session:key')
+  assert.deepEqual(cache.start('session:key', secondHash, 102), { status: 'started' })
+})
+
+test('expires stale entries lazily', () => {
+  const cache = new ImportIdempotencyCache<string>(10)
+  const firstHash = hashImportBody({ messages: ['first'] })
+  const secondHash = hashImportBody({ messages: ['second'] })
+
+  assert.deepEqual(cache.start('session:key', firstHash, 100), { status: 'started' })
+  assert.deepEqual(cache.start('session:key', secondHash, 111), { status: 'started' })
+})

packages/node-runtime/src/services/import-idempotency.ts

@@ -0,0 +1,61 @@
+import { createHash } from 'node:crypto'
+
+export const DEFAULT_IMPORT_IDEMPOTENCY_TTL_MS = 60 * 60 * 1000
+
+interface ImportIdempotencyEntry<T> {
+  bodyHash: string
+  status: 'pending' | 'success'
+  response?: T
+  timestamp: number
+}
+
+export type ImportIdempotencyStartResult<T> =
+  | { status: 'started' }
+  | { status: 'conflict' }
+  | { status: 'pending' }
+  | { status: 'success'; response: T }
+
+export function hashImportBody(body: unknown): string {
+  return createHash('sha256')
+    .update(JSON.stringify(body ?? ''))
+    .digest('hex')
+}
+
+/**
+ * 进程内的请求级幂等缓存。调用方先 start，再在业务完成后 success/fail，
+ * 确保 Desktop、CLI 和 Internal Server 使用同一套冲突与重试语义。
+ */
+export class ImportIdempotencyCache<T> {
+  private entries = new Map<string, ImportIdempotencyEntry<T>>()
+
+  constructor(private ttlMs = DEFAULT_IMPORT_IDEMPOTENCY_TTL_MS) {}
+
+  start(key: string, bodyHash: string, now = Date.now()): ImportIdempotencyStartResult<T> {
+    this.prune(now)
+    const entry = this.entries.get(key)
+    if (!entry) {
+      this.entries.set(key, { bodyHash, status: 'pending', timestamp: now })
+      return { status: 'started' }
+    }
+    if (entry.bodyHash !== bodyHash) return { status: 'conflict' }
+    if (entry.status === 'pending') return { status: 'pending' }
+    return { status: 'success', response: entry.response as T }
+  }
+
+  success(key: string, response: T): void {
+    const entry = this.entries.get(key)
+    if (!entry) return
+    entry.status = 'success'
+    entry.response = response
+  }
+
+  fail(key: string): void {
+    this.entries.delete(key)
+  }
+
+  private prune(now: number): void {
+    for (const [key, entry] of this.entries) {
+      if (now - entry.timestamp > this.ttlMs) this.entries.delete(key)
+    }
+  }
+}

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

@@ -81,8 +81,9 @@ export type {
 export { MergeSessionCache } from './merge-cache'
 
 // Push import (POST /api/v1/imports/:sessionId)
-export { pushImport } from './push-importer'
+export { executePushImportUnlocked, pushImport } from './push-importer'
 export type {
+  PushImportExecutionDeps,
   PushImportPayload,
   PushImportResult,
   PushImportOutcome,
@@ -90,3 +91,6 @@ export type {
   PushImportMember,
   PushImportMeta,
 } from './push-importer'
+
+export { DEFAULT_IMPORT_IDEMPOTENCY_TTL_MS, hashImportBody, ImportIdempotencyCache } from './import-idempotency'
+export type { ImportIdempotencyStartResult } from './import-idempotency'