@@ -172,9 +172,15 @@ async function stageBundleBytes(importId: string, bytes: Buffer): Promise<string
 }
 
 async function runImport(importId: string, sourceKey: string, name: string) {
-  await runKnowledgeImport({
+  await runKnowledgeImport(await importPayload(importId, sourceKey, name))
+  const record = await findKnowledgeImport({ importId, assertedWorkspaceId: ids.workspaceId })
+  if (!record) throw new Error('Import record disappeared')
+  return record
+}
+
+async function importPayload(importId: string, sourceKey: string, name: string) {
+  return {
     importId,
-    knowledgeBaseId: generateId(),
     workspaceId: ids.workspaceId,
     userId: ids.aliceId,
     name,
@@ -183,10 +189,7 @@ async function runImport(importId: string, sourceKey: string, name: string) {
       actorUserId: ids.aliceId,
       workspaceId: ids.workspaceId,
     }),
-  })
-  const record = await findKnowledgeImport({ importId, assertedWorkspaceId: ids.workspaceId })
-  if (!record) throw new Error('Import record disappeared')
-  return record
+  }
 }
 
 async function importedDocuments(knowledgeBaseId: string) {
@@ -482,6 +485,27 @@ describe('knowledge base bundle import', () => {
     ).rejects.toThrow()
   })
 
+  it('lets exactly one of two concurrent workers run an import', async () => {
+    const importId = generateId()
+    const sourceKey = await stageExportedBundle(importId)
+    const payload = await importPayload(importId, sourceKey, 'Claimed fixture')
+
+    await Promise.all([runKnowledgeImport(payload), runKnowledgeImport(payload)])
+
+    expect(await findKnowledgeImport({ importId })).toMatchObject({
+      status: 'completed',
+      knowledgeBaseId: importId,
+      documentsImported: 2,
+      documentsFailed: 0,
+    })
+    expect(
+      await db
+        .select({ id: knowledgeBase.id })
+        .from(knowledgeBase)
+        .where(eq(knowledgeBase.name, 'Claimed fixture'))
+    ).toEqual([{ id: importId }])
+  })
+
   it('fails without a partial base when the name is already taken', async () => {
     const importId = generateId()
     const sourceKey = await stageExportedBundle(importId)
@@ -583,6 +607,62 @@ describe('knowledge base import requests', () => {
       .toEqual([{ resourceId: finished.knowledgeBaseId }])
   })
 
+  it('starts an import whose first dispatch was lost when its completion is retried', async () => {
+    const bytes = await exportedBundleBytes()
+    const created = await createKnowledgeImport.execute({
+      principal: session,
+      input: {
+        body: {
+          workspaceId: ids.workspaceId,
+          name: 'Redispatched import',
+          source: {
+            type: 'upload',
+            name: 'fixture.simkb.zip',
+            contentType: 'application/zip',
+            size: bytes.length,
+          },
+        },
+      },
+      request,
+    })
+    await writeLocalPutObject({
+      uploadId: created.upload.id,
+      key: created.upload.finalKey,
+      body: new Blob([new Uint8Array(bytes)]).stream(),
+      expectedSize: bytes.length,
+      contentType: created.upload.contentType,
+      metadata: uploadSessionObjectMetadata(created.upload),
+    })
+    // The record a completion wrote before the request died, with no worker behind it.
+    await insertKnowledgeImport({
+      id: created.upload.id,
+      workspaceId: ids.workspaceId,
+      userId: ids.aliceId,
+      sourceKey: created.upload.storageKey,
+    })
+    const input = {
+      importId: created.import.id,
+      workspaceId: ids.workspaceId,
+      uploadToken: created.upload.uploadToken,
+    }
+
+    await completeKnowledgeImport.execute({ principal: session, input, request })
+
+    await expect
+      .poll(
+        async () =>
+          (await readKnowledgeImport.execute({ principal: session, input })).import.status,
+        { interval: 50, timeout: 10_000 }
+      )
+      .toBe('completed')
+    expect(
+      await db
+        .select({ id: knowledgeBase.id })
+        .from(knowledgeBase)
+        .where(eq(knowledgeBase.name, 'Redispatched import'))
+    ).toEqual([{ id: created.import.id }])
+  })
+
   it('refuses a bundle that is not a zip archive before creating an upload', async () => {
     const importUploads = () =>
       db
 
@@ -223,6 +223,24 @@ function findStartedImport(context: KnowledgeImportUploadContext) {
   })
 }
 
+/** What the runner needs, with usage admitted for the caller completing the upload. */
+async function importPayload(
+  principal: Principal,
+  context: KnowledgeImportUploadContext
+): Promise<KnowledgeImportPayload> {
+  const { billingAttribution, userId } = await assertKnowledgeUsageAdmitted(principal, context)
+  const { upload } = context
+  const name = importedName(upload)
+  return {
+    importId: upload.id,
+    workspaceId: context.workspaceId,
+    userId,
+    ...(name ? { name } : {}),
+    sourceKey: upload.storageKey,
+    billingAttribution,
+  }
+}
+
 /**
  * Writes the durable record while the upload is being completed, before it
  * counts as complete. From then on the stale-import janitor owns the bundle, so
@@ -235,34 +253,27 @@ async function recordKnowledgeImport(
   principal: Principal,
   context: KnowledgeImportUploadContext
 ): Promise<KnowledgeImportPayload | null> {
-  const { billingAttribution, userId } = await assertKnowledgeUsageAdmitted(principal, context)
-  const { upload } = context
+  const payload = await importPayload(principal, context)
   const inserted = await insertKnowledgeImport({
-    id: upload.id,
-    workspaceId: context.workspaceId,
-    userId,
-    sourceKey: upload.storageKey,
+    id: payload.importId,
+    workspaceId: payload.workspaceId,
+    userId: payload.userId,
+    sourceKey: payload.sourceKey,
+  })
+  return inserted ? payload : null
+}
+
+function startKnowledgeImport(payload: KnowledgeImportPayload): Promise<void> {
+  return dispatchBackgroundTask<typeof knowledgeImportTask>('knowledge-import', payload, {
+    runInProcess: runKnowledgeImport,
+    trigger: { tags: [`workspaceId:${payload.workspaceId}`, `importId:${payload.importId}`] },
   })
-  if (!inserted) return null
-  const name = importedName(upload)
-  return {
-    importId: upload.id,
-    knowledgeBaseId: generateId(),
-    workspaceId: context.workspaceId,
-    userId,
-    ...(name ? { name } : {}),
-    sourceKey: upload.storageKey,
-    billingAttribution,
-  }
 }
 
 /** Starts the runner; a dispatch that fails settles the import and drops its bundle. */
 async function dispatchKnowledgeImport(payload: KnowledgeImportPayload): Promise<void> {
   try {
-    await dispatchBackgroundTask<typeof knowledgeImportTask>('knowledge-import', payload, {
-      runInProcess: runKnowledgeImport,
-      trigger: { tags: [`workspaceId:${payload.workspaceId}`, `importId:${payload.importId}`] },
-    })
+    await startKnowledgeImport(payload)
   } catch (error) {
     await finishKnowledgeImport(payload.importId, {
       status: 'failed',
@@ -280,6 +291,29 @@ async function dispatchKnowledgeImport(payload: KnowledgeImportPayload): Promise
   }
 }
 
+/**
+ * Dispatches again an import whose record no worker has claimed: its first
+ * dispatch was lost with the request that wrote the record, or is still
+ * queued. Either way a second run is safe, since only one worker can claim it,
+ * and a failure here leaves the import to the stale-import janitor rather than
+ * failing one that may already be on its way.
+ */
+async function redispatchUnclaimedImport(
+  principal: Principal,
+  context: KnowledgeImportUploadContext,
+  record: KnowledgeImportRecord
+): Promise<void> {
+  if (record.status !== 'running' || record.startedAt) return
+  try {
+    await startKnowledgeImport(await importPayload(principal, context))
+  } catch (error) {
+    logger.warn('Could not dispatch an unclaimed knowledge import again', {
+      importId: record.id,
+      error: getErrorMessage(error, 'Unknown error'),
+    })
+  }
+}
+
 export const createKnowledgeImport = defineAuthorizedKnowledgeUseCase({
   operation: knowledgeOperations.importCreate,
   resolveContext: ({ input }: { input: CreateKnowledgeImportInput }) =>
@@ -367,7 +401,10 @@ export const completeKnowledgeImport = defineAuthorizedKnowledgeUseCase({
   }) => resolveUploadContext(principal, input),
   async execute({ principal, context }): Promise<CompleteKnowledgeImportResult> {
     const existing = await findStartedImport(context)
-    if (existing) return { import: fromRecord(existing), startedKnowledgeBaseId: null }
+    if (existing) {
+      await redispatchUnclaimedImport(principal, context, existing)
+      return { import: fromRecord(existing), startedKnowledgeBaseId: null }
+    }
     let started: KnowledgeImportPayload | null
     try {
       const completed = await completeUploadSession({
@@ -405,7 +442,7 @@ export const completeKnowledgeImport = defineAuthorizedKnowledgeUseCase({
       importId: context.upload.id,
       assertedWorkspaceId: context.workspaceId,
     })
-    return { import: fromRecord(record), startedKnowledgeBaseId: started?.knowledgeBaseId ?? null }
+    return { import: fromRecord(record), startedKnowledgeBaseId: started?.importId ?? null }
   },
   /** Recorded once per import: a repeated completion only reads back the one already started. */
   projectAudit: ({ context, result }) =>
 
@@ -8,7 +8,7 @@
 import { db } from '@sim/db'
 import { knowledgeBaseImport } from '@sim/db/schema'
 import { truncate } from '@sim/utils/string'
-import { and, eq } from 'drizzle-orm'
+import { and, eq, isNull } from 'drizzle-orm'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
 
 /** Where an import's uploaded bundle is stored until the import settles. */
@@ -73,6 +73,20 @@ export async function getKnowledgeImport(params: {
   return record
 }
 
+/**
+ * Claims a running import for one worker. Returns `false` when another worker
+ * already claimed it, or it settled, so a duplicate dispatch exits untouched.
+ */
+export async function claimKnowledgeImport(importId: string): Promise<boolean> {
+  const now = new Date()
+  const claimed = await db
+    .update(knowledgeBaseImport)
+    .set({ startedAt: now, updatedAt: now })
+    .where(and(whileRunning(importId), isNull(knowledgeBaseImport.startedAt)))
+    .returning({ id: knowledgeBaseImport.id })
+  return claimed.length > 0
+}
+
 /**
  * Records the runner's progress, which doubles as its heartbeat. Returns
  * `false` once the import is no longer running, which tells the runner to stop.
 
@@ -30,6 +30,8 @@ import {
   openKnowledgeBundle,
 } from '@/lib/knowledge/transfer/import-archive'
 import {
+  claimKnowledgeImport,
+  findKnowledgeImport,
   finishKnowledgeImport,
   KNOWLEDGE_IMPORT_STORAGE_CONTEXT,
   type KnowledgeImportFailure,
@@ -50,10 +52,13 @@ const logger = createLogger('KnowledgeImportRunner')
  */
 const DOCUMENT_CONCURRENCY = 2
 
+/**
+ * What a worker needs to run an import. The base it creates takes the import's
+ * id, so every dispatch of one import names the same base, and the import's
+ * audit names it before it exists.
+ */
 export interface KnowledgeImportPayload {
   importId: string
-  /** Minted when the upload completed, so the audit of the import names this base. */
-  knowledgeBaseId: string
   workspaceId: string
   /** The person the base and its documents are attributed to. */
   userId: string
@@ -239,7 +244,7 @@ async function importBundle(
   if (!(await updateRunningKnowledgeImport(payload.importId, {}))) return
   const knowledgeBase = await createAuthorizedKnowledgeBase(
     {
-      id: payload.knowledgeBaseId,
+      id: payload.importId,
       name: payload.name ?? manifest.knowledgeBase.name,
       description: manifest.knowledgeBase.description ?? undefined,
       workspaceId: payload.workspaceId,
@@ -262,16 +267,22 @@ async function importBundle(
     return
   }
 
-  for (const tag of manifest.tags) {
-    await createTagDefinition(
-      {
-        knowledgeBaseId: knowledgeBase.id,
-        tagSlot: tag.slot,
-        displayName: tag.displayName,
-        fieldType: tag.fieldType,
-      },
-      requestId
-    )
+  try {
+    for (const tag of manifest.tags) {
+      await createTagDefinition(
+        {
+          knowledgeBaseId: knowledgeBase.id,
+          tagSlot: tag.slot,
+          displayName: tag.displayName,
+          fieldType: tag.fieldType,
+        },
+        requestId
+      )
+    }
+  } catch (error) {
+    // No document has landed yet, so a base without its tags is removed rather than kept half-built.
+    await deleteKnowledgeBase(knowledgeBase.id, requestId)
+    throw error
   }
 
   const context: ImportDocumentContext = {
@@ -332,11 +343,18 @@ async function importBundle(
 }
 
 /**
- * Runs one import to a settled state. The bundle is copied to local disk so
- * its entries can be read by offset, and both copies are removed however the
- * import ends.
+ * Runs one import to a settled state. The worker claims the import first, so a
+ * second dispatch of it leaves the first worker's bundle alone. The bundle is
+ * copied to local disk so its entries can be read by offset, and both copies
+ * are removed however the import ends.
  */
 export async function runKnowledgeImport(payload: KnowledgeImportPayload): Promise<void> {
+  if (!(await claimKnowledgeImport(payload.importId))) {
+    // Settled before any worker claimed it, so no worker is reading the bundle.
+    const record = await findKnowledgeImport({ importId: payload.importId })
+    if (record && !record.startedAt) await deleteBundle(payload)
+    return
+  }
   const directory = await mkdtemp(join(tmpdir(), 'knowledge-import-'))
   let reader: KnowledgeBundleReader | undefined
   try {
@@ -365,13 +383,17 @@ export async function runKnowledgeImport(payload: KnowledgeImportPayload): Promi
   } finally {
     reader?.close()
     await rm(directory, { recursive: true, force: true })
-    await deleteFile({ key: payload.sourceKey, context: KNOWLEDGE_IMPORT_STORAGE_CONTEXT }).catch(
-      (error: unknown) => {
-        logger.warn('Failed to delete an imported knowledge bundle', {
-          importId: payload.importId,
-          error: getErrorMessage(error, 'Unknown error'),
-        })
-      }
-    )
+    await deleteBundle(payload)
   }
 }
+
+async function deleteBundle(payload: KnowledgeImportPayload): Promise<void> {
+  await deleteFile({ key: payload.sourceKey, context: KNOWLEDGE_IMPORT_STORAGE_CONTEXT }).catch(
+    (error: unknown) => {
+      logger.warn('Failed to delete an imported knowledge bundle', {
+        importId: payload.importId,
+        error: getErrorMessage(error, 'Unknown error'),
+      })
+    }
+  )
+}
@@ -13,6 +13,7 @@ CREATE TABLE "knowledge_base_import" (
 	"failures" jsonb DEFAULT '[]'::jsonb NOT NULL,
 	"error" text,
 	"created_at" timestamp DEFAULT now() NOT NULL,
+	"started_at" timestamp,
 	"updated_at" timestamp DEFAULT now() NOT NULL,
 	"completed_at" timestamp
 );