fix(kb): keep the collection when a file is indexed after assignment (#1200)

Assigning a collection before indexing lost it silently. The per-row
value still showed, but Manage Collections and the Search in dropdown
stayed empty, because getKnowledgeCollections() facets on the Qdrant
payload while only MySQL had been written.

Five gaps on one path:

updateFileCollection() sets the payload filtered on `source`, which
matches nothing before the file is indexed. It also only persisted to
kb_ingest_state `if (row)`, so a file with no row stored the value
nowhere at all and still returned "Moved to ...".

Six of the seven EmbedFileJob.dispatch sites never pass `collection`,
and none read the existing row, so Index dispatched a job with no
knowledge of the assignment. The ZIM branch of processAndEmbedFile then
dropped `collection` even when the job had one, so ZIM content could
never be tagged at embed time by any path. Batch continuations dropped
it too, which would have tagged only batch 1.

Resolve the effective collection once inside EmbedFileJob.handle rather
than at seven call sites, thread it through the ZIM path into the point
payload, carry it across batch continuations, and make the pre-index
assignment durable with getOrCreate.

Verified end to end on a test appliance: assigned a collection to an
unindexed ZIM, indexed it across multiple batch continuations, and all
6106 chunks carry the tag.

admin/app/jobs/embed_file_job.ts

@@ -59,6 +59,15 @@ export class EmbedFileJob {
   async handle(job: Job) {
     const { filePath, fileName, batchOffset, totalArticles, collection } = job.data as EmbedFileJobParams
 
+    // Only the direct KB-upload controller passes `collection` on dispatch; the other
+    // six dispatch sites (download auto-index, scan/sync, re-embed, local ZIM upload,
+    // replaced-file reconcile, and this job's own ZIM batch continuation) do not. Fall
+    // back to whatever the file is already assigned to, so an assignment made *before*
+    // the file was indexed still reaches the vectors. Resolving it here rather than at
+    // each dispatch site keeps one source of truth and covers batch continuations too.
+    const effectiveCollection =
+      collection ?? (await KbIngestState.findBy('file_path', filePath))?.collection ?? undefined
+
     const isZimBatch = batchOffset !== undefined
     const batchInfo = isZimBatch ? ` (batch offset: ${batchOffset})` : ''
     logger.info(`[EmbedFileJob] Starting embedding process for: ${fileName}${batchInfo}`)
@@ -138,7 +147,7 @@ export class EmbedFileJob {
         allowDeletion,
         batchOffset,
         onProgress,
-        collection
+        effectiveCollection
       )
 
       if (!result.success) {
@@ -192,6 +201,9 @@ export class EmbedFileJob {
           totalArticles: totalArticles || result.totalArticles,
           isFinalBatch: false, // Explicitly not final
           chunksSoFar: chunksSoFarNext,
+          // Carry the collection across batches, otherwise only batch 1 of a ZIM
+          // would be tagged and the rest would land uncategorized.
+          ...(effectiveCollection ? { collection: effectiveCollection } : {}),
         })
 
         // Calculate progress based on articles processed.
@@ -244,7 +256,7 @@ export class EmbedFileJob {
       // BullMQ's :completed retention (50 jobs) ages out, so the state row is
       // the only durable record of "this file finished embedding".
       try {
-        await KbIngestState.markIndexed(filePath, totalChunks, collection)
+        await KbIngestState.markIndexed(filePath, totalChunks, effectiveCollection)
       } catch (stateErr) {
         logger.warn(
           `[EmbedFileJob] Failed to persist ingest state for ${fileName}: %s`,

admin/app/services/rag_service.ts

@@ -529,7 +529,8 @@ export class RagService {
     filepath: string,
     deleteAfterEmbedding: boolean,
     batchOffset?: number,
-    onProgress?: (percent: number) => Promise<void>
+    onProgress?: (percent: number) => Promise<void>,
+    collection?: string
   ): Promise<ProcessZIMFileResponse> {
     const zimExtractionService = new ZIMExtractionService()
 
@@ -556,6 +557,9 @@ export class RagService {
       const result = await this.embedAndStoreText(zimChunk.text, {
         source: filepath,
         content_type: 'zim_article',
+        // Without this the ZIM path writes points with no `collection` at all, so
+        // getKnowledgeCollections() (which facets on it) never sees them.
+        ...(collection ? { collection } : {}),
 
         // Article-level context
         article_title: zimChunk.articleTitle,
@@ -783,7 +787,7 @@ export class RagService {
       // Process based on file type
       // ZIM files are handled specially since they have their own embedding workflow
       if (fileType === 'zim') {
-        return await this.processZIMFile(filepath, deleteAfterEmbedding, batchOffset, onProgress)
+        return await this.processZIMFile(filepath, deleteAfterEmbedding, batchOffset, onProgress, collection)
       }
 
       // Extract text based on file type
@@ -1240,11 +1244,13 @@ export class RagService {
         filter: { must: [{ key: 'source', match: { value: source } }] },
       })
 
-      const row = await KbIngestState.query().where('file_path', source).first()
-      if (row) {
-        row.collection = collection
-        await row.save()
-      }
+      // The setPayload above only reaches points that already exist, so for a file
+      // that has not been indexed yet it matches nothing. The row is therefore the
+      // only durable record of the choice until EmbedFileJob picks it up -- create
+      // it when absent rather than reporting success and storing the value nowhere.
+      const row = await KbIngestState.getOrCreate(source)
+      row.collection = collection
+      await row.save()
 
       return { success: true, message: collection ? `Moved to "${collection}".` : 'Moved to Uncategorized.' }
     } catch (error) {