feat(KB): per-file ingest state machine (Phase 1 of RFC #883) (#888)

Adds a persistent state machine for AI knowledge-base ingestion so the
scanner can distinguish "fully indexed", "user opted out", "failed", and
"stalled" from each other — none of which were derivable from the prior
binary "any chunks in Qdrant ⇒ embedded" check.

## What lands

- New table `kb_ingest_state` keyed by `file_path` with enum state column
  (`pending_decision | indexed | browse_only | failed | stalled`).
  Independent of `installed_resources` so it covers both curated downloads
  and manually-uploaded KB files.
- New KV key `rag.defaultIngestPolicy` (string: `Always | Manual`).
  Registered now but not consumed yet — JIT prompt + wizard step land in
  Phase 3 of the RFC.
- `EmbedFileJob.handle` writes state on terminal outcomes:
  - Success (final batch) → `indexed` + chunks count
  - `UnrecoverableError` → `failed` + error message
  - Retryable errors are left to BullMQ's existing retry path
- `scanAndSyncStorage` swaps the binary qdrant check for a state-aware
  decision tree (see `decideScanAction`). Existing installs auto-backfill
  on first scan: files with chunks in Qdrant but no state row become
  `indexed`; new files start as `pending_decision`.
- `deleteFileBySource` drops the state row last, so removed files
  disappear entirely instead of leaving an orphan that the next scan
  would re-dispatch into nothing.

## What does NOT land here

- Ratio registry (separate PR) — needed for partial-stall detection and
  cost estimates, but a separable concern.
- #880 follow-up initial-progress anchor (separate tiny PR).
- Phase 2 UI (status pill, per-card actions, conditional warnings).
- Phase 3 policy surfaces (wizard step, JIT prompt, guardrail modal).
- PR #886's bulk-action hookup — `_deletePointsBySource` / Re-embed All
  / Reset & Rebuild would also want to set state, but #886 isn't merged
  yet; that wiring goes in a follow-up once #886 lands.

## Target

This is forward work for v1.40.0 (RFC #883). Branching off `rc` because
that's the current latest base and post-GA Jake will sync rc→dev; a
retarget at PR-open time is a fast-forward if requested.

## Tests

- 9 new unit tests for `decideScanAction` covering all five states plus
  the no-row / chunks-present / chunks-missing combinations
- Type-check clean
- Smoke-tested end-to-end on NOMAD3 via hot-patch:
  - Backfill: 5 ZIMs + 2 KB uploads with existing chunks in Qdrant all
    came back `indexed` on first scan
  - Pending dispatch: a video-only ZIM with no chunks (`lrnselfreliance`)
    came back `pending_decision` and was correctly re-dispatched (Bull
    deduped to its historical `:completed` jobId — bgauger's #886 fix
    drains that)
  - Delete hook: deleting a KB upload via `DELETE /api/rag/files`
    removed both the disk file and the state row

Co-authored-by: Jake Turner <52841588+jakeaturner@users.noreply.github.com>

admin/app/jobs/embed_file_job.ts

@@ -4,6 +4,7 @@ import { EmbedJobWithProgress } from '../../types/rag.js'
 import { RagService } from '#services/rag_service'
 import { DockerService } from '#services/docker_service'
 import { OllamaService } from '#services/ollama_service'
+import KbIngestState from '#models/kb_ingest_state'
 import { createHash } from 'crypto'
 import logger from '@adonisjs/core/services/logger'
 import fs from 'node:fs/promises'
@@ -193,6 +194,18 @@ export class EmbedFileJob {
         chunks: totalChunks,
       })
 
+      // Persist the post-job state so scanAndSyncStorage knows this file is done.
+      // 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)
+      } catch (stateErr) {
+        logger.warn(
+          `[EmbedFileJob] Failed to persist ingest state for ${fileName}: %s`,
+          stateErr instanceof Error ? stateErr.message : String(stateErr)
+        )
+      }
+
       const batchMsg = isZimBatch ? ` (final batch, total chunks: ${totalChunks})` : ''
       logger.info(
         `[EmbedFileJob] Successfully embedded ${result.chunks} chunks from file: ${fileName}${batchMsg}`
@@ -215,6 +228,23 @@ export class EmbedFileJob {
         error: error instanceof Error ? error.message : 'Unknown error',
       })
 
+      // Only persist `failed` for unrecoverable errors. Retryable errors get
+      // automatic BullMQ retries (30 attempts); marking state failed on every
+      // transient blip would suppress the retry-driven recovery path.
+      if (error instanceof UnrecoverableError) {
+        try {
+          await KbIngestState.markFailed(
+            filePath,
+            error instanceof Error ? error.message : 'Unknown error'
+          )
+        } catch (stateErr) {
+          logger.warn(
+            `[EmbedFileJob] Failed to persist failed state for ${fileName}: %s`,
+            stateErr instanceof Error ? stateErr.message : String(stateErr)
+          )
+        }
+      }
+
       throw error
     }
   }

admin/app/models/kb_ingest_state.ts

@@ -0,0 +1,77 @@
+import { DateTime } from 'luxon'
+import { BaseModel, column, SnakeCaseNamingStrategy } from '@adonisjs/lucid/orm'
+import type { KbIngestStateValue } from '../../types/kb_ingest_state.js'
+
+const LAST_ERROR_MAX_LEN = 1024
+
+/**
+ * Tracks the per-file decision and outcome of AI knowledge-base ingestion.
+ *
+ * The row exists for any embeddable file the scanner has seen and is independent
+ * of `installed_resources` (which only covers curated downloads). Replaces the
+ * earlier "any chunks in qdrant ⇒ embedded" binary check, which conflated
+ * partially-stalled ingestions with fully-indexed files. See RFC #883.
+ */
+export default class KbIngestState extends BaseModel {
+  static table = 'kb_ingest_state'
+  static namingStrategy = new SnakeCaseNamingStrategy()
+
+  @column({ isPrimary: true })
+  declare id: number
+
+  @column()
+  declare file_path: string
+
+  @column()
+  declare state: KbIngestStateValue
+
+  @column()
+  declare chunks_embedded: number
+
+  @column()
+  declare last_error: string | null
+
+  @column.dateTime({ autoCreate: true })
+  declare created_at: DateTime
+
+  @column.dateTime({ autoCreate: true, autoUpdate: true })
+  declare updated_at: DateTime
+
+  static async getOrCreate(filePath: string): Promise<KbIngestState> {
+    return this.firstOrCreate(
+      { file_path: filePath },
+      { file_path: filePath, state: 'pending_decision', chunks_embedded: 0 }
+    )
+  }
+
+  static async markIndexed(filePath: string, chunksEmbedded: number): Promise<void> {
+    const row = await this.getOrCreate(filePath)
+    row.state = 'indexed'
+    row.chunks_embedded = chunksEmbedded
+    row.last_error = null
+    await row.save()
+  }
+
+  static async markFailed(filePath: string, errorMessage: string): Promise<void> {
+    const row = await this.getOrCreate(filePath)
+    row.state = 'failed'
+    row.last_error = errorMessage.slice(0, LAST_ERROR_MAX_LEN)
+    await row.save()
+  }
+
+  static async markBrowseOnly(filePath: string): Promise<void> {
+    const row = await this.getOrCreate(filePath)
+    row.state = 'browse_only'
+    await row.save()
+  }
+
+  static async markStalled(filePath: string): Promise<void> {
+    const row = await this.getOrCreate(filePath)
+    row.state = 'stalled'
+    await row.save()
+  }
+
+  static async remove(filePath: string): Promise<void> {
+    await this.query().where('file_path', filePath).delete()
+  }
+}

admin/app/services/rag_service.ts

@@ -16,6 +16,8 @@ import { removeStopwords } from 'stopword'
 import { randomUUID } from 'node:crypto'
 import { join, resolve, sep } from 'node:path'
 import KVStore from '#models/kv_store'
+import KbIngestState from '#models/kb_ingest_state'
+import { decideScanAction } from '../utils/kb_ingest_decision.js'
 import { ZIMExtractionService } from './zim_extraction_service.js'
 import { ZIM_BATCH_SIZE } from '../../constants/zim_extraction.js'
 import { ProcessAndEmbedFileResponse, ProcessZIMFileResponse, RAGResult, RerankedRAGResult } from '../../types/rag.js'
@@ -1118,6 +1120,11 @@ export class RagService {
         logger.warn(`[RAG] File was removed from knowledge base but doesn't live in Nomad's uploads directory, so it can't be safely removed. Skipping deletion of physical file...`)
       }
 
+      // Drop the ingest state row last so the file disappears entirely. Without
+      // this, the next scanAndSyncStorage would see `indexed + no chunks` for a
+      // path that no longer exists in storage and try to re-embed nothing.
+      await KbIngestState.remove(source)
+
       return { success: true, message: 'File removed from knowledge base.' }
     } catch (error) {
       logger.error('[RAG] Error deleting file from knowledge base:', error)
@@ -1330,8 +1337,64 @@ export class RagService {
         offset = scrollResult.next_page_offset || null
       } while (offset !== null)
 
-      const filesToEmbed = filesInStorage.filter((f) => !sourcesInQdrant.has(f))
-      logger.info(`[RAG] ${filesToEmbed.length} of ${filesInStorage.length} files need embedding`)
+      logger.info(`[RAG] Found ${sourcesInQdrant.size} unique sources in Qdrant`)
+
+      // Load all known per-file ingest states. The state row is authoritative
+      // over the "any chunks in Qdrant" heuristic — it captures user choices
+      // (browse_only) and terminal outcomes (failed, stalled) that aren't visible
+      // from Qdrant alone. See RFC #883 for the full state machine.
+      const stateRows = await KbIngestState.all()
+      const stateByPath = new Map(stateRows.map((row) => [row.file_path, row]))
+
+      // Non-embeddable files (e.g. kiwix-library.xml in /storage/zim) would otherwise
+      // be dispatched to EmbedFileJob, fail with "Unsupported file type", and retry
+      // on every sync — filter them out before state decisions.
+      const embeddableFiles = filesInStorage.filter(
+        (filePath) => determineFileType(filePath) !== 'unknown'
+      )
+
+      const filesToEmbed: string[] = []
+      let backfilled = 0
+      let createdRows = 0
+      let skipped = 0
+
+      for (const filePath of embeddableFiles) {
+        const stateRow = stateByPath.get(filePath) ?? null
+        const action = decideScanAction(stateRow, sourcesInQdrant.has(filePath))
+
+        switch (action.kind) {
+          case 'skip':
+            skipped++
+            break
+          case 'backfill_indexed':
+            // Pre-RFC install (or a fresh admin pointed at an existing Qdrant volume):
+            // chunks already exist with no state row, so trust Qdrant and record
+            // `indexed` without re-embedding. chunks_embedded is left 0 because
+            // we don't count points-per-source during the scroll above.
+            await KbIngestState.create({
+              file_path: filePath,
+              state: 'indexed',
+              chunks_embedded: 0,
+            })
+            backfilled++
+            break
+          case 'dispatch':
+            if (action.createStateRow) {
+              await KbIngestState.create({
+                file_path: filePath,
+                state: 'pending_decision',
+                chunks_embedded: 0,
+              })
+              createdRows++
+            }
+            filesToEmbed.push(filePath)
+            break
+        }
+      }
+
+      logger.info(
+        `[RAG] Scan results: ${filesToEmbed.length} to embed, ${backfilled} backfilled, ${createdRows} new pending, ${skipped} skipped`
+      )
 
       if (filesToEmbed.length === 0) {
         return {

admin/app/utils/kb_ingest_decision.ts

@@ -0,0 +1,53 @@
+import type { KbIngestStateValue } from '../../types/kb_ingest_state.js'
+
+/**
+ * Decision returned by `decideScanAction` describing what scanAndSyncStorage
+ * should do for one file given its current state row (if any) and whether
+ * Qdrant already has chunks for it.
+ *
+ * - `skip` — file is in a settled state (already indexed, deliberately not
+ *   indexed, or in a manual-recovery state); no auto-dispatch.
+ * - `dispatch` — file needs to be (re-)embedded; an EmbedFileJob should be
+ *   dispatched. `createStateRow` indicates whether a new state row needs to
+ *   be created before dispatch (i.e. first time the scanner has seen it).
+ * - `backfill_indexed` — Qdrant has chunks but no state row exists yet
+ *   (pre-RFC install, or new admin instance pointed at an existing Qdrant
+ *   volume). Create a row in `indexed` state without re-embedding.
+ */
+export type ScanAction =
+  | { kind: 'skip' }
+  | { kind: 'dispatch'; createStateRow: boolean }
+  | { kind: 'backfill_indexed' }
+
+export interface KbIngestStateRow {
+  state: KbIngestStateValue
+}
+
+/**
+ * Decide what scanAndSyncStorage should do for a single embeddable file.
+ *
+ * Replaces the earlier `!sourcesInQdrant.has(filePath)` binary check, which
+ * couldn't tell a fully-indexed file from a stalled mid-batch ingestion, and
+ * couldn't honor a user's "browse only" choice. The state row is now the
+ * authoritative answer; Qdrant chunk presence is corroborating evidence.
+ */
+export function decideScanAction(
+  stateRow: KbIngestStateRow | null,
+  hasChunksInQdrant: boolean
+): ScanAction {
+  if (!stateRow) {
+    if (hasChunksInQdrant) return { kind: 'backfill_indexed' }
+    return { kind: 'dispatch', createStateRow: true }
+  }
+
+  switch (stateRow.state) {
+    case 'indexed':
+      return hasChunksInQdrant ? { kind: 'skip' } : { kind: 'dispatch', createStateRow: false }
+    case 'pending_decision':
+      return { kind: 'dispatch', createStateRow: false }
+    case 'browse_only':
+    case 'failed':
+    case 'stalled':
+      return { kind: 'skip' }
+  }
+}

admin/database/migrations/1776000000001_create_kb_ingest_state_table.ts

@@ -0,0 +1,26 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+export default class extends BaseSchema {
+  protected tableName = 'kb_ingest_state'
+
+  async up() {
+    this.schema.createTable(this.tableName, (table) => {
+      table.increments('id').primary()
+      // utf8mb4 caps an indexed varchar at 768 chars (3072 byte InnoDB key limit);
+      // 512 leaves headroom and is plenty for any NOMAD-managed file path.
+      table.string('file_path', 512).notNullable().unique()
+      table
+        .enum('state', ['pending_decision', 'indexed', 'browse_only', 'failed', 'stalled'])
+        .notNullable()
+        .defaultTo('pending_decision')
+      table.integer('chunks_embedded').notNullable().defaultTo(0)
+      table.text('last_error').nullable()
+      table.timestamp('created_at').notNullable()
+      table.timestamp('updated_at').notNullable()
+    })
+  }
+
+  async down() {
+    this.schema.dropTable(this.tableName)
+  }
+}

admin/tests/unit/kb_ingest_decision.spec.ts

@@ -0,0 +1,46 @@
+import * as assert from 'node:assert/strict'
+import { test } from 'node:test'
+
+import { decideScanAction } from '../../app/utils/kb_ingest_decision.js'
+
+test('no state row, no chunks → dispatch and create row (new file)', () => {
+  assert.deepEqual(decideScanAction(null, false), { kind: 'dispatch', createStateRow: true })
+})
+
+test('no state row, chunks present → backfill_indexed (pre-RFC install, existing Qdrant volume)', () => {
+  assert.deepEqual(decideScanAction(null, true), { kind: 'backfill_indexed' })
+})
+
+test('indexed + chunks present → skip', () => {
+  assert.deepEqual(decideScanAction({ state: 'indexed' }, true), { kind: 'skip' })
+})
+
+test('indexed + chunks missing → re-dispatch (state stale, Qdrant collection rebuilt or chunks deleted)', () => {
+  assert.deepEqual(decideScanAction({ state: 'indexed' }, false), {
+    kind: 'dispatch',
+    createStateRow: false,
+  })
+})
+
+test('pending_decision → dispatch (preserves current Always behavior until policy is consumed)', () => {
+  assert.deepEqual(decideScanAction({ state: 'pending_decision' }, false), {
+    kind: 'dispatch',
+    createStateRow: false,
+  })
+})
+
+test('browse_only → skip (user opted out of indexing)', () => {
+  assert.deepEqual(decideScanAction({ state: 'browse_only' }, false), { kind: 'skip' })
+})
+
+test('browse_only + chunks present → skip (do not silently re-index after un-index)', () => {
+  assert.deepEqual(decideScanAction({ state: 'browse_only' }, true), { kind: 'skip' })
+})
+
+test('failed → skip (manual retry needed, do not auto-redispatch)', () => {
+  assert.deepEqual(decideScanAction({ state: 'failed' }, false), { kind: 'skip' })
+})
+
+test('stalled → skip (manual retry needed)', () => {
+  assert.deepEqual(decideScanAction({ state: 'stalled' }, false), { kind: 'skip' })
+})

admin/types/kb_ingest_state.ts

@@ -0,0 +1,9 @@
+export const KB_INGEST_STATES = [
+  'pending_decision',
+  'indexed',
+  'browse_only',
+  'failed',
+  'stalled',
+] as const
+
+export type KbIngestStateValue = (typeof KB_INGEST_STATES)[number]

admin/types/kv_store.ts

@@ -3,6 +3,7 @@ export const KV_STORE_SCHEMA = {
   'chat.suggestionsEnabled':    'boolean',
   'chat.lastModel':             'string',
   'rag.docsEmbedded':           'boolean',
+  'rag.defaultIngestPolicy':    'string',
   'system.updateAvailable':     'boolean',
   'system.latestVersion':       'string',
   'system.earlyAccess':         'boolean',