feat(kb): implement shared ingestion executor and simplify delete operations

Backend changes:
- Add shared thread pool _ingest_executor to prevent global thread pool exhaustion
- Implement graceful shutdown with timeout mechanism (30s) for ingestion executor
- Fix rename_collection_api cross-user data pollution by adding user_id filter
- Improve rename_collection_api transaction safety with pre-validation
- Simplify delete_document_api: remove complex UploadedFile and physical file handling
- Physical file cleanup is now handled at collection level only

Frontend changes:
- Extend IngestionResult interface with embedding_count, vector_count, warnings, failed_step
- Add toast warning when embedding_count is 0 but chunks_count > 0
- Improve error handling for both document and web ingestion with partial status warnings

Author: sqhyz55

frontend/src/components/kb/knowledge-base-creation-dialog.tsx

@@ -33,8 +33,12 @@ interface IngestionResult {
   collection: string
   document_count: number
   chunks_count: number
+  embedding_count?: number
+  vector_count?: number
   status: string
   message: string
+  warnings?: string[]
+  failed_step?: string
 }
 
 interface WebIngestionResult {
@@ -287,13 +291,24 @@ export function KnowledgeBaseCreationDialog({ open, onOpenChange, onSuccess }: K
           throw new Error(errorData.detail || t("kb.errors.uploadFailedFile", { name: file.name }))
         }
 
-        const result = await response.json()
-        setIngestionResults(prev => [...prev, result])
+        const result: IngestionResult = await response.json()
 
-        if (result.status === "partial" && result.failed_step) {
-          throw new Error(result.message || t("kb.errors.failedAtStep", { step: result.failed_step }))
+        if (result.status === "error") {
+          throw new Error(result.message || t("kb.errors.uploadFailedFile", { name: file.name }))
+        }
+        if (result.status === "partial") {
+          toast.warning(
+            result.message + (result.warnings?.length ? ` Warnings: ${result.warnings.join(", ")}` : "")
+          )
+        }
+        if (result.status === "success" && (result.embedding_count ?? 0) === 0 && (result.chunks_count ?? 0) > 0) {
+          toast.error(
+            "文档上传成功，但 embedding 生成失败（" + result.chunks_count + " 个 chunks 未生成 embedding）。文档无法被搜索。请检查 embedding 模型配置和 API 状态。" +
+            (result.warnings?.length ? ` 警告: ${result.warnings.join(", ")}` : "")
+          )
         }
 
+        setIngestionResults(prev => [...prev, result])
         successfulCollections.push(collectionName)
         setUploadProgress(((i + 1) / selectedFiles.length) * 100)
       }
@@ -370,6 +385,16 @@ export function KnowledgeBaseCreationDialog({ open, onOpenChange, onSuccess }: K
       }
 
       const result: WebIngestionResult = await response.json()
+
+      if (result.status === "error") {
+        throw new Error(result.message || t("kb.errors.webIngestFailed"))
+      }
+      if (result.status === "partial") {
+        toast.warning(
+          result.message + (result.warnings?.length ? ` Warnings: ${result.warnings.join(", ")}` : "")
+        )
+      }
+
       setWebIngestionResult(result)
       setWebIngestionProgress(100)
 

src/xagent/web/api/kb.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import concurrent.futures
+import contextvars
 import functools
 import json
 import logging
@@ -102,6 +103,50 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
 # Create router
 kb_router = APIRouter(prefix="/api/kb", tags=["kb"])
 
+# Shared executor for ingestion tasks to prevent global thread pool exhaustion
+_ingest_executor = concurrent.futures.ThreadPoolExecutor(
+    max_workers=10, thread_name_prefix="ingest_worker_"
+)
+
+
+def shutdown_ingest_executor() -> None:
+    """Shutdown the shared ingestion executor gracefully.
+
+    Called during application shutdown to ensure pending tasks complete
+    and resources are properly released. Uses a timeout to prevent blocking
+    application shutdown indefinitely.
+    """
+    logger.info("Shutting down ingestion executor...")
+    try:
+        # Use threading.Event to implement timeout with executor shutdown
+        import threading
+
+        shutdown_complete = threading.Event()
+
+        def wait_for_shutdown() -> None:
+            _ingest_executor.shutdown(wait=True)
+            shutdown_complete.set()
+
+        shutdown_thread = threading.Thread(target=wait_for_shutdown, daemon=True)
+        shutdown_thread.start()
+
+        # Wait for shutdown with timeout
+        if shutdown_complete.wait(timeout=30):
+            logger.info("Ingestion executor shutdown complete")
+        else:
+            logger.warning(
+                "Executor shutdown timed out after 30s; forcing shutdown. "
+                "Some ingestion tasks may be incomplete."
+            )
+            _ingest_executor.shutdown(wait=False)
+    except Exception as e:
+        logger.error("Error during executor shutdown: %s", e)
+        # Force shutdown on any error
+        try:
+            _ingest_executor.shutdown(wait=False)
+        except Exception:
+            pass  # Ignore errors during forced shutdown
+
 
 class CloudFile(BaseModel):
     provider: str
@@ -427,19 +472,18 @@ async def ingest(
 
     progress_manager = get_progress_manager()
 
-    def _run_ingestion() -> IngestionResult:
-        return run_document_ingestion(
+    result: IngestionResult = await asyncio.get_running_loop().run_in_executor(
+        _ingest_executor,
+        lambda: contextvars.copy_context().run(
+            run_document_ingestion,
             collection=collection,
             source_path=str(file_path),
             ingestion_config=config,
             progress_manager=progress_manager,
             user_id=int(_user.id),
             is_admin=bool(_user.is_admin),
-        )
-
-    with concurrent.futures.ThreadPoolExecutor() as executor:
-        future = executor.submit(_run_ingestion)
-        result: IngestionResult = future.result()
+        ),
+    )
 
     if result.status == "error":
         return JSONResponse(status_code=500, content=result.model_dump())
@@ -545,14 +589,17 @@ def _download_file() -> None:
 
                     # Run ingestion (blocking)
                     try:
-                        result = await asyncio.to_thread(
-                            run_document_ingestion,
-                            collection=request.collection,
-                            source_path=str(file_path),
-                            ingestion_config=config,
-                            progress_manager=progress_manager,
-                            user_id=int(_user.id),
-                            is_admin=bool(_user.is_admin),
+                        result = await asyncio.get_running_loop().run_in_executor(
+                            _ingest_executor,
+                            lambda: contextvars.copy_context().run(
+                                run_document_ingestion,
+                                collection=request.collection,
+                                source_path=str(file_path),
+                                ingestion_config=config,
+                                progress_manager=progress_manager,
+                                user_id=int(_user.id),
+                                is_admin=bool(_user.is_admin),
+                            ),
                         )
                         return result
                     except Exception as e:
@@ -864,15 +911,16 @@ async def ingest_web(
         max_retries: Maximum retry attempts
         retry_delay: Delay between retries
     """
+    # SECURITY: Validate collection name at API boundary
     try:
-        try:
-            safe_collection = sanitize_path_component(collection, "collection")
-        except ValueError as e:
-            logger.warning("Invalid collection name rejected: %s - %s", collection, e)
-            raise HTTPException(
-                status_code=422, detail=f"Invalid collection name: {str(e)}"
-            ) from e
+        safe_collection = sanitize_path_component(collection, "collection")
+    except ValueError as e:
+        logger.warning("Invalid collection name rejected: %s - %s", collection, e)
+        raise HTTPException(
+            status_code=422, detail=f"Invalid collection name: {str(e)}"
+        ) from e
 
+    try:
         url_patterns_list = (
             [p.strip() for p in url_patterns.split(",")] if url_patterns else None
         )
@@ -958,16 +1006,18 @@ async def ingest_web(
             ),
         )
 
-        result = await asyncio.get_event_loop().run_in_executor(
-            None,
-            lambda: asyncio.run(
+        # Run web ingestion
+        result = await asyncio.get_running_loop().run_in_executor(
+            _ingest_executor,
+            lambda: contextvars.copy_context().run(
+                asyncio.run,
                 run_web_ingestion(
                     collection=safe_collection,
                     crawl_config=crawl_config,
                     ingestion_config=ingestion_config,
                     user_id=int(_user.id),
                     is_admin=bool(_user.is_admin),
-                )
+                ),
             ),
         )
 
@@ -1321,6 +1371,10 @@ async def delete_document_api(
         This endpoint uses filename lookup which may have a race condition if
         the same filename is uploaded multiple times concurrently. For production
         use, consider using doc_id directly or adding a filename index column.
+
+        Physical file cleanup is handled at the collection level (when the entire
+        collection is deleted). This endpoint only removes the document from
+        LanceDB vector storage.
     """
     # NOTE: Exceptions are normalized by @handle_kb_exceptions for consistent API responses.
     from ...core.tools.core.RAG_tools.LanceDB.schema_manager import (
@@ -1642,33 +1696,66 @@ async def rename_collection_api(
         physical_rename_error = f"Path resolution error: {str(e)}"
 
     # Step 2: Update collection name in all tables
+    # NOTE: LanceDB does not support ACID transactions across multiple tables.
+    # We use a best-effort approach: validate all tables are accessible first,
+    # then perform updates. If any update fails, we record the failure but cannot
+    # roll back previously completed updates.
     table_names = _list_table_names(conn, warnings)
 
+    # Pre-flight validation: ensure all target tables are accessible
+    tables_to_update = []
     for table_name in ["documents", "parses", "chunks"]:
         if table_name in table_names:
-            try:
-                table = conn.open_table(table_name)
-                table.update(
-                    f"collection = '{escape_lancedb_string(collection_name)}'",
-                    {"collection": new_name},
-                )
-            except Exception as e:
-                logger.warning("Failed to update '%s': %s", table_name, e)
-                warnings.append(f"Failed to update '{table_name}': {e}")
+            tables_to_update.append(table_name)
 
     for table_name in table_names:
-        if not table_name.startswith("embeddings_"):
-            continue
+        if table_name.startswith("embeddings_"):
+            tables_to_update.append(table_name)
+
+    # Validate all tables can be opened
+    validation_errors = []
+    for table_name in tables_to_update:
+        try:
+            conn.open_table(table_name)
+        except Exception as e:
+            validation_errors.append(f"Cannot access table '{table_name}': {e}")
+
+    if validation_errors:
+        # Fail fast if any table is inaccessible
+        error_msg = f"Cannot rename collection: {', '.join(validation_errors)}"
+        logger.error("Rename validation failed: %s", error_msg)
+        raise HTTPException(
+            status_code=500,
+            detail=error_msg,
+        )
+
+    # Perform all updates (best-effort, no rollback)
+    update_success = {}
+    update_failed = {}
+
+    for table_name in tables_to_update:
         try:
             table = conn.open_table(table_name)
             table.update(
-                f"collection = '{escape_lancedb_string(collection_name)}'",
+                f"collection = '{escape_lancedb_string(collection_name)}' AND user_id = {int(_user.id)}",
                 {"collection": new_name},
             )
+            update_success[table_name] = True
         except Exception as e:
-            logger.warning("Failed to update embeddings table '%s': %s", table_name, e)
+            update_failed[table_name] = str(e)
+            logger.error("Failed to update table '%s': %s", table_name, e)
             warnings.append(f"Failed to update '{table_name}': {e}")
 
+    # Log summary of update results
+    if update_failed:
+        logger.warning(
+            "Collection rename update summary: %d succeeded, %d failed. Success: %s, Failed: %s",
+            len(update_success),
+            len(update_failed),
+            list(update_success.keys()),
+            list(update_failed.keys()),
+        )
+
     # Migrate ingestion status from old collection name to new
     try:
         status_entries = load_ingestion_status(collection=collection_name)
@@ -1734,11 +1821,13 @@ async def rename_collection_api(
             "status": final_status,
             "message": final_message,
             "warnings": warnings,
+            "physical_move": physical_rename_status,
         }
 
     return {
         "status": "success",
         "message": f"Collection renamed from '{collection_name}' to '{new_name}'",
+        "physical_move": physical_rename_status,
     }
 
 

src/xagent/web/app.py

@@ -212,6 +212,12 @@ async def shutdown_event() -> None:
     if sandbox_mgr:
         await sandbox_mgr.cleanup()
 
+    # Shutdown shared ingestion executor
+    from .api.kb import shutdown_ingest_executor
+
+    shutdown_ingest_executor()
+    logger.info("Ingestion executor shutdown completed")
+
 
 # Frontend is now served by Next.js at http://localhost:3000
 # This backend only provides API endpoints