feat(RAG): adds the ability to cancel all embedding jobs (#1034)

admin/app/controllers/rag_controller.ts

@@ -110,6 +110,14 @@ export default class RagController {
     })
   }
 
+  public async cancelAllJobs({ response }: HttpContext) {
+    const result = await EmbedFileJob.cancelAllJobs()
+    return response.status(200).json({
+      message: `Cancelled ${result.cancelled} job${result.cancelled !== 1 ? 's' : ''}${result.filesDeleted > 0 ? `, deleted ${result.filesDeleted} file${result.filesDeleted !== 1 ? 's' : ''}` : ''}.`,
+      ...result,
+    })
+  }
+
   public async policyPromptState({ response }: HttpContext) {
     const result = await this.ragService.getPolicyPromptState()
     return response.status(200).json(result)

admin/app/jobs/embed_file_job.ts

@@ -164,6 +164,22 @@ export class EmbedFileJob {
           await new Promise((resolve) => setTimeout(resolve, EmbedFileJob.CPU_BATCH_DELAY_MS))
         }
 
+        // Bail before re-populating the queue if this job was cancelled mid-batch.
+        // cancelAllJobs() obliterates the queue (including this active job), but a
+        // worker already inside handle() would otherwise dispatch its continuation
+        // afterwards and silently revive a cancelled ZIM ingestion. If our own job
+        // key is gone, the cancel happened — skip the dispatch. Mirrors the
+        // "tolerate external removal" handling in safeUpdateProgress above.
+        const stillQueued = await QueueService.getInstance()
+          .getQueue(EmbedFileJob.queue)
+          .getJob(job.id!)
+        if (!stillQueued) {
+          logger.info(
+            `[EmbedFileJob] Job ${fileName} was cancelled; skipping continuation dispatch`
+          )
+          return { success: false, cancelled: true, fileName, filePath }
+        }
+
         // Dispatch next batch (not final yet). Carry forward the running
         // chunk count so the final batch can persist an accurate total (#933).
         const chunksSoFarNext = (job.data.chunksSoFar || 0) + (result.chunks || 0)
@@ -443,6 +459,46 @@ export class EmbedFileJob {
     return { cleaned, filesDeleted }
   }
 
+  /** Unconditionally clear every embedding job regardless of state.
+   *
+   *  cleanupFailedJobs only removes jobs explicitly tagged status === 'failed',
+   *  which leaves stuck jobs (waiting / active / delayed / paused that never
+   *  reached 'failed') unreachable from the UI — the operator's only recourse was
+   *  flushing Redis by hand. This wipes the whole queue, including a locked active
+   *  job, via obliterate({ force: true }) (plain obliterate/job.remove throw on a
+   *  locked job). It touches only Redis, so it is safe while Qdrant/Ollama are
+   *  offline — which is exactly when jobs pile up and wedge. */
+  static async cancelAllJobs(): Promise<{ cancelled: number; filesDeleted: number }> {
+    const queueService = QueueService.getInstance()
+    const queue = queueService.getQueue(this.queue)
+    const jobs = await queue.getJobs(['waiting', 'active', 'delayed', 'paused', 'failed'])
+
+    let filesDeleted = 0
+    for (const job of jobs) {
+      const filePath = (job.data as EmbedFileJobParams).filePath
+      // Same guard as cleanupFailedJobs: only delete user uploads, never ZIM
+      // library files or Nomad docs that live outside the uploads path.
+      if (filePath && filePath.includes(RagService.UPLOADS_STORAGE_PATH)) {
+        try {
+          await fs.unlink(filePath)
+          filesDeleted++
+        } catch {
+          // File may already be deleted — that's fine
+        }
+      }
+    }
+
+    const cancelled = jobs.length
+
+    // force: true removes the locked/active job too. An in-flight worker may keep
+    // running its current batch in memory; the self-exists guard in handle()
+    // prevents it from dispatching a continuation back into the cleared queue.
+    await queue.obliterate({ force: true })
+
+    logger.info(`[EmbedFileJob] Cancelled ${cancelled} jobs, deleted ${filesDeleted} files`)
+    return { cancelled, filesDeleted }
+  }
+
   static async getStatus(filePath: string): Promise<{
     exists: boolean
     status?: string

admin/inertia/components/chat/KnowledgeBaseModal.tsx

@@ -262,6 +262,20 @@ export default function KnowledgeBaseModal({ aiAssistantName = "AI Assistant", o
     },
   })
 
+  const cancelAllMutation = useMutation({
+    mutationFn: () => api.cancelAllEmbedJobs(),
+    onSuccess: (data) => {
+      addNotification({ type: 'success', message: data?.message || 'All embedding jobs cancelled.' })
+      queryClient.invalidateQueries({ queryKey: ['embed-jobs'] })
+      queryClient.invalidateQueries({ queryKey: ['failedEmbedJobs'] })
+      queryClient.invalidateQueries({ queryKey: ['storedFiles'] })
+      queryClient.invalidateQueries({ queryKey: ['kbFileWarnings'] })
+    },
+    onError: (error: any) => {
+      addNotification({ type: 'error', message: error?.message || 'Failed to cancel jobs.' })
+    },
+  })
+
   const startQdrantMutation = useMutation({
     mutationFn: () => api.affectService(SERVICE_NAMES.QDRANT, 'start'),
     onSuccess: () => {
@@ -358,6 +372,31 @@ export default function KnowledgeBaseModal({ aiAssistantName = "AI Assistant", o
     }
   }
 
+  const handleConfirmCancelAll = () => {
+    openModal(
+      <StyledModal
+        title='Cancel All Embedding Jobs?'
+        onConfirm={() => {
+          cancelAllMutation.mutate()
+          closeModal('confirm-cancel-all-modal')
+        }}
+        onCancel={() => closeModal('confirm-cancel-all-modal')}
+        open={true}
+        confirmText='Cancel All Jobs'
+        cancelText='Keep Jobs'
+        confirmVariant='danger'
+      >
+        <p className='text-text-primary'>
+          This stops <strong>every</strong> embedding job — including ones still in progress or
+          stuck — and clears the processing queue. The uploaded source files for those jobs are
+          deleted, so you'll need to re-upload anything you still want indexed. Stored files that
+          already finished embedding are not affected. Are you sure you want to proceed?
+        </p>
+      </StyledModal>,
+      'confirm-cancel-all-modal'
+    )
+  }
+
   const handleConfirmSync = () => {
     openModal(
       <StyledModal
@@ -533,18 +572,33 @@ export default function KnowledgeBaseModal({ aiAssistantName = "AI Assistant", o
           </div>
 
           <div className="my-8">
-            <div className="flex items-center justify-between mb-4">
+            <div className="flex items-center justify-between mb-4 gap-2 flex-wrap">
               <StyledSectionHeader title="Processing Queue" className="!mb-0" />
-              <StyledButton
-                variant="danger"
-                size="md"
-                icon="IconTrash"
-                onClick={() => cleanupFailedMutation.mutate()}
-                loading={cleanupFailedMutation.isPending}
-                disabled={cleanupFailedMutation.isPending || qdrantOffline}
-              >
-                Clean Up Failed
-              </StyledButton>
+              <div className="flex items-center gap-2 flex-wrap">
+                <StyledButton
+                  variant="danger"
+                  size="md"
+                  icon="IconTrash"
+                  onClick={() => cleanupFailedMutation.mutate()}
+                  loading={cleanupFailedMutation.isPending}
+                  disabled={cleanupFailedMutation.isPending || qdrantOffline}
+                >
+                  Clean Up Failed
+                </StyledButton>
+                {/* Not gated on qdrantOffline: clearing stuck jobs must work during
+                    a Qdrant/Ollama outage, which is exactly when they wedge. */}
+                <StyledButton
+                  variant="danger"
+                  size="md"
+                  icon="IconPlayerStop"
+                  onClick={handleConfirmCancelAll}
+                  loading={cancelAllMutation.isPending}
+                  disabled={cancelAllMutation.isPending}
+                  title="Stop and clear every embedding job regardless of state, including stuck or in-progress ones. Deletes the uploaded source files for those jobs."
+                >
+                  Cancel All Jobs
+                </StyledButton>
+              </div>
             </div>
             <ActiveEmbedJobs withHeader={false} />
           </div>

admin/inertia/lib/api.ts

@@ -479,6 +479,13 @@ class API {
     })()
   }
 
+  async cancelAllEmbedJobs(): Promise<{ message: string; cancelled: number; filesDeleted: number } | undefined> {
+    return catchInternal(async () => {
+      const response = await this.client.delete<{ message: string; cancelled: number; filesDeleted: number }>('/rag/jobs')
+      return response.data
+    })()
+  }
+
   async checkRAGHealth() {
     return catchInternal(async () => {
       const response = await this.client.get<{ online: boolean; message?: string }>('/rag/health')

admin/start/routes.ts

@@ -153,6 +153,7 @@ router
     router.get('/active-jobs', [RagController, 'getActiveJobs'])
     router.get('/failed-jobs', [RagController, 'getFailedJobs'])
     router.delete('/failed-jobs', [RagController, 'cleanupFailedJobs'])
+    router.delete('/jobs', [RagController, 'cancelAllJobs'])
     router.get('/job-status', [RagController, 'getJobStatus'])
     router.post('/sync', [RagController, 'scanAndSync'])
     router.post('/re-embed-all', [RagController, 'reembedAll'])