feat: add support for skipped tasks and warnings in task processing

ricofurtado · ricofurtado · commit bf2072b12a94 · 2026-06-05T03:36:42.000-04:00
diff --git a/frontend/app/api/queries/useGetTasksQuery.ts b/frontend/app/api/queries/useGetTasksQuery.ts
@@ -29,6 +29,7 @@ export interface TaskFileEntry {
     | "running"
     | "processing"
     | "completed"
+    | "skipped"
     | "failed"
     | "error";
   result?: unknown;
@@ -57,6 +58,7 @@ export interface Task {
     | "running"
     | "processing"
     | "completed"
+    | "skipped"
     | "failed"
     | "error";
   total_files?: number;
diff --git a/frontend/components/task-dialog/constants.ts b/frontend/components/task-dialog/constants.ts
@@ -1,4 +1,10 @@
-import { AlertCircle, CheckCircle, Clock, type LucideIcon } from "lucide-react";
+import {
+  AlertCircle,
+  CheckCircle,
+  Clock,
+  type LucideIcon,
+  TriangleAlert,
+} from "lucide-react";
 import type { TaskFileStatusCategory } from "@/lib/task-utils";
 
 export const CATEGORY_CHIPS: Array<{
@@ -19,6 +25,12 @@ export const CATEGORY_CHIPS: Array<{
     icon: AlertCircle,
     iconClassName: "text-destructive",
   },
+  {
+    id: "warning",
+    label: "Warning",
+    icon: TriangleAlert,
+    iconClassName: "text-brand-amber",
+  },
   {
     id: "indexing",
     label: "Indexing",
diff --git a/frontend/components/task-error-content.tsx b/frontend/components/task-error-content.tsx
@@ -17,10 +17,12 @@ import {
 } from "@/lib/task-error-display";
 import {
   getFailedFileCount,
-  getFailedFileEntries,
   getSuccessfulFileCount,
   getTaskFileName,
+  getTaskIssueFileEntries,
+  getWarningFileEntries,
   isCompletedTotalFailure,
+  isTaskFileWarning,
   isTerminalFailedTask,
 } from "@/lib/task-utils";
 import { formatTaskTimestamp, parseTimestamp } from "@/lib/time-utils";
@@ -48,15 +50,20 @@ export function TaskErrorContent({
   );
   const isExpanded = accordionValue === "failed-files";
 
-  const failedEntries = useMemo(() => getFailedFileEntries(task), [task]);
+  const issueEntries = useMemo(() => getTaskIssueFileEntries(task), [task]);
 
   const failedCount = getFailedFileCount(task);
+  const warningCount = getWarningFileEntries(task).length;
   const successCount = getSuccessfulFileCount(task);
   const timestamp =
     parseTimestamp(task.created_at) ?? parseTimestamp(task.updated_at);
   const isFailedStatus =
     isTerminalFailedTask(task) || isCompletedTotalFailure(task);
-  const statusLabel = isFailedStatus ? "Failed" : "Complete";
+  const statusLabel = isFailedStatus
+    ? "Failed"
+    : warningCount > 0
+      ? "Warning"
+      : "Complete";
   // Pill colors: failed (red) vs partial success (amber/orange), each with IBM tokens or OSS borders.
   const statusPillClassName = cn(
     "shrink-0 rounded-full px-2 py-1 text-xs",
@@ -69,7 +76,7 @@ export function TaskErrorContent({
         : "border border-brand-amber-30 bg-brand-amber-10 text-brand-amber",
   );
 
-  if (failedCount <= 0 && failedEntries.length === 0) {
+  if (failedCount <= 0 && issueEntries.length === 0) {
     return null;
   }
 
@@ -78,7 +85,9 @@ export function TaskErrorContent({
   const accordionSummary = (
     <div className="flex min-w-0 flex-1 items-center gap-1">
       <span className="text-xs">
-        {successCount} success · {failedCount} failed
+        {successCount} success
+        {warningCount > 0 ? ` · ${warningCount} warning` : ""}
+        {failedCount > 0 ? ` · ${failedCount} failed` : ""}
       </span>
       <ChevronDown className="size-4 shrink-0 transition-transform group-data-[state=open]:rotate-180" />
     </div>
@@ -174,19 +183,30 @@ export function TaskErrorContent({
             {accordionHeader}
             <AccordionContent className="w-full p-0 pt-2">
               <div className="flex w-full flex-col gap-2">
-                {failedEntries.map(([filePath, fileInfo], index) => {
+                {issueEntries.map(([filePath, fileInfo], index) => {
                   const fileName = getTaskFileName(filePath, fileInfo);
                   const line = resolveTaskFileError(fileInfo, task.error);
                   const componentCause = formatApiComponent(fileInfo.component);
+                  const isWarning = isTaskFileWarning(fileInfo);
 
                   return (
                     <div
                       key={`${task.task_id}-${filePath}-${index}`}
                       className={cn(
                         "task-failed-file-card min-w-0",
                         isCloudBrand
-                          ? "flex flex-col items-start gap-2 self-stretch rounded-none rounded-r border-l-[1.5px] border-l-destructive bg-border p-2"
-                          : "flex flex-col gap-1 rounded border-destructive/20 bg-failure-soft py-mmd px-4",
+                          ? cn(
+                              "flex flex-col items-start gap-2 self-stretch rounded-none rounded-r border-l-[1.5px] bg-border p-2",
+                              isWarning
+                                ? "border-l-brand-amber"
+                                : "border-l-destructive",
+                            )
+                          : cn(
+                              "flex flex-col gap-1 rounded py-mmd px-4",
+                              isWarning
+                                ? "border border-brand-amber-30 bg-brand-amber-10"
+                                : "border-destructive/20 bg-failure-soft",
+                            ),
                       )}
                     >
                       <p
diff --git a/frontend/lib/task-error-display.ts b/frontend/lib/task-error-display.ts
@@ -137,6 +137,15 @@ export function resolveTaskFileError(
   fileInfo: TaskFileEntry,
   taskError?: string,
 ): string {
+  if (
+    fileInfo.result &&
+    typeof fileInfo.result === "object" &&
+    "warning" in fileInfo.result &&
+    typeof fileInfo.result.warning === "string" &&
+    fileInfo.result.warning.trim()
+  ) {
+    return fileInfo.result.warning.trim();
+  }
   if (typeof fileInfo.user_facing_message === "string") {
     const message = fileInfo.user_facing_message.trim();
     if (message) {
diff --git a/frontend/lib/task-utils.ts b/frontend/lib/task-utils.ts
@@ -7,7 +7,11 @@ import {
 export const ALL_TASK_FILE_TYPES = "__all__";
 export const ALL_TASK_STATUS_CATEGORIES = "__all__";
 
-export type TaskFileStatusCategory = "completed" | "system_error" | "indexing";
+export type TaskFileStatusCategory =
+  | "completed"
+  | "warning"
+  | "system_error"
+  | "indexing";
 
 export type TaskFileNameSort = "asc" | "desc";
 
@@ -26,6 +30,10 @@ export function isTaskFileFailed(fileInfo: TaskFileEntry): boolean {
   return fileInfo.status === "failed" || fileInfo.status === "error";
 }
 
+export function isTaskFileWarning(fileInfo: TaskFileEntry): boolean {
+  return fileInfo.status === "skipped";
+}
+
 export function getTaskFileDialogStatusLabel(
   fileInfo: TaskFileEntry,
   taskError?: string,
@@ -40,6 +48,9 @@ export function getTaskFileDialogStatusLabel(
     }
     return buildRowStatusLabel("unknown");
   }
+  if (isTaskFileWarning(fileInfo)) {
+    return "Warning";
+  }
   if (isTaskFileCompleted(fileInfo)) {
     return "Complete";
   }
@@ -126,6 +137,9 @@ export function getTaskFileStatusCategory(
   if (isTaskFileFailed(fileInfo)) {
     return "system_error";
   }
+  if (isTaskFileWarning(fileInfo)) {
+    return "warning";
+  }
 
   const status = fileInfo.status ?? "pending";
   if (status === "pending" || status === "running" || status === "processing") {
@@ -144,6 +158,7 @@ export function countTaskFileEntriesByCategory(
 ): Record<TaskFileStatusCategory, number> {
   const counts: Record<TaskFileStatusCategory, number> = {
     completed: 0,
+    warning: 0,
     system_error: 0,
     indexing: 0,
   };
@@ -218,11 +233,27 @@ export function getFailedFileEntries(
   );
 }
 
+export function getWarningFileEntries(
+  task: Task,
+): Array<[string, TaskFileEntry]> {
+  return Object.entries(task.files || {}).filter(([, fileInfo]) =>
+    isTaskFileWarning(fileInfo),
+  );
+}
+
+export function getTaskIssueFileEntries(
+  task: Task,
+): Array<[string, TaskFileEntry]> {
+  return Object.entries(task.files || {}).filter(
+    ([, fileInfo]) => isTaskFileFailed(fileInfo) || isTaskFileWarning(fileInfo),
+  );
+}
+
 export function hasFailedFileEntries(task: Task): boolean {
   if ((task.failed_files ?? 0) > 0) {
     return true;
   }
-  return getFailedFileEntries(task).length > 0;
+  return getTaskIssueFileEntries(task).length > 0;
 }
 
 export function isTerminalFailedTask(task: Task): boolean {
diff --git a/src/models/processors.py b/src/models/processors.py
@@ -741,12 +741,21 @@ async def process_item(self, upload_task: UploadTask, item: str, file_task: File
             )
             if await self.check_filename_exists(document.filename, opensearch_client):
                 if not self.replace_duplicates:
-                    file_task.status = TaskStatus.FAILED
-                    file_task.error = f"File with name '{document.filename}' already exists"
+                    file_task.status = TaskStatus.SKIPPED
+                    file_task.error = None
+                    file_task.result = {
+                        "status": "skipped",
+                        "reason": "duplicate_filename",
+                        "warning": "A file with this name already exists.",
+                    }
                     file_task.updated_at = time.time()
-                    upload_task.failed_files += 1
+                    upload_task.successful_files += 1
                     return
-                await self.delete_document_by_filename(document.filename, opensearch_client)
+                await self.delete_document_by_filename(
+                    document.filename,
+                    opensearch_client,
+                    owner_user_id=self.user_id,
+                )
 
             # Create temporary file from document content
             suffix = os.path.splitext(document.filename)[1]
diff --git a/tests/unit/test_connector_processor_filename_dedupe.py b/tests/unit/test_connector_processor_filename_dedupe.py
@@ -96,7 +96,7 @@ async def mock_search(index, body, **kwargs):
 
 
 @pytest.mark.asyncio
-async def test_connector_processor_fails_when_filename_exists_and_replace_false(monkeypatch):
+async def test_connector_processor_skips_when_filename_exists_and_replace_false(monkeypatch):
     monkeypatch.setattr("config.settings.DISABLE_INGEST_WITH_LANGFLOW", True)
     processor = _build_connector_processor(replace_duplicates=False)
     document = _make_document()
@@ -108,10 +108,15 @@ async def test_connector_processor_fails_when_filename_exists_and_replace_false(
     with patch.object(processor, "process_document_standard", new=AsyncMock()) as mock_process:
         await processor.process_item(upload_task, "file-id-1", file_task)
 
-    assert file_task.status == TaskStatus.FAILED
-    assert "already exists" in (file_task.error or "")
-    assert upload_task.failed_files == 1
-    assert upload_task.successful_files == 0
+    assert file_task.status == TaskStatus.SKIPPED
+    assert file_task.error is None
+    assert file_task.result == {
+        "status": "skipped",
+        "reason": "duplicate_filename",
+        "warning": "A file with this name already exists.",
+    }
+    assert upload_task.failed_files == 0
+    assert upload_task.successful_files == 1
     mock_process.assert_not_called()
     opensearch_client.delete_by_query.assert_not_called()
 
@@ -271,7 +276,7 @@ async def mock_search(index, body, **kwargs):
 
 
 @pytest.mark.asyncio
-async def test_langflow_connector_processor_fails_on_filename_collision(monkeypatch):
+async def test_langflow_connector_processor_skips_on_filename_collision(monkeypatch):
     monkeypatch.setattr("config.settings.DISABLE_INGEST_WITH_LANGFLOW", False)
     processor = _build_langflow_processor(replace_duplicates=False)
     document = _make_document()
@@ -282,9 +287,15 @@ async def test_langflow_connector_processor_fails_on_filename_collision(monkeypa
 
     await processor.process_item(upload_task, "file-id-1", file_task)
 
-    assert file_task.status == TaskStatus.FAILED
-    assert "already exists" in (file_task.error or "")
-    assert upload_task.failed_files == 1
+    assert file_task.status == TaskStatus.SKIPPED
+    assert file_task.error is None
+    assert file_task.result == {
+        "status": "skipped",
+        "reason": "duplicate_filename",
+        "warning": "A file with this name already exists.",
+    }
+    assert upload_task.failed_files == 0
+    assert upload_task.successful_files == 1
     processor.connector_service.langflow_service.upload_and_ingest_file.assert_not_called()
     opensearch_client.delete_by_query.assert_not_called()
 
