refac

backend/open_webui/routers/files.py

@@ -41,7 +41,7 @@ from open_webui.models.knowledge import Knowledges
 from open_webui.models.users import Users
 from open_webui.retrieval.vector.async_client import ASYNC_VECTOR_DB_CLIENT
 from open_webui.routers.audio import transcribe
-from open_webui.routers.retrieval import ProcessFileForm, process_file
+from open_webui.routers.retrieval import ProcessFileForm, index_file, process_file
 from open_webui.storage.provider import Storage
 from open_webui.utils.auth import get_admin_user, get_verified_user
 from open_webui.utils.misc import strict_match_mime_type
@@ -136,6 +136,7 @@ async def process_uploaded_file(
 ):
     async def _process_handler(db_session):
         try:
+            await Files.update_file_data_by_id(file_item.id, {'status': 'processing'}, db=db_session)
             content_type = file.content_type
 
             # Detect mis-labeled text files (e.g. .ts → video/mp2t)
@@ -158,7 +159,7 @@ async def process_uploaded_file(
                     file_metadata,
                     user,
                 )
-                await process_file(
+                await index_file(
                     request,
                     ProcessFileForm(file_id=file_item.id, content=result.get('text', '')),
                     user=user,
@@ -178,11 +179,6 @@ async def process_uploaded_file(
                     # extraction causes "Timeout reached while detecting
                     # encoding" errors.
                     log.info('Video file detected (%s), skipping text extraction', content_type)
-                    await Files.update_file_data_by_id(
-                        file_item.id,
-                        {'status': 'completed'},
-                        db=db_session,
-                    )
                 else:
                     raise Exception(f'File type {content_type} is not supported for processing')
 
@@ -191,7 +187,7 @@ async def process_uploaded_file(
                 # configured content extraction engine.
                 if not content_type:
                     log.info('File type %s is not provided, but trying to process anyway', file.content_type)
-                await process_file(
+                await index_file(
                     request,
                     ProcessFileForm(file_id=file_item.id),
                     user=user,
@@ -233,10 +229,7 @@ async def process_uploaded_file(
                                 )
                                 directory_id = None
 
-                        # Keep the generic file status stream open until the
-                        # KB-specific vector write and durable link both finish.
-                        await Files.update_file_data_by_id(file_item.id, {'status': 'processing'}, db=db_session)
-                        await process_file(
+                        await index_file(
                             request,
                             ProcessFileForm(file_id=file_item.id, collection_name=knowledge_id),
                             user=user,
@@ -256,6 +249,9 @@ async def process_uploaded_file(
                     log.warning(f'Failed to link file {file_item.id} to knowledge {knowledge_id}: {e}')
                     raise
 
+            # Completion includes the durable knowledge link, when requested.
+            await Files.update_file_data_by_id(file_item.id, {'status': 'completed', 'error': None}, db=db_session)
+
         except Exception as e:
             log.error(f'Error processing file: {file_item.id}')
             await Files.update_file_data_by_id(

backend/open_webui/routers/retrieval.py

@@ -1575,8 +1575,26 @@ async def process_file(
     user=Depends(get_verified_user),
     db: AsyncSession = Depends(get_async_session),
 ):
+    """Standalone processing endpoint: index the file, then mark it completed.
+
+    Uploads call index_file directly so they can finish knowledge linking before
+    reporting completion.
     """
-    Process a file and save its content to the vector database.
+    result = await index_file(request, form_data, user, db)
+    async with get_async_db() as session:
+        await Files.update_file_data_by_id(form_data.file_id, {'status': 'completed', 'error': None}, db=session)
+    return result
+
+
+async def index_file(
+    request: Request,
+    form_data: ProcessFileForm,
+    user,
+    db: AsyncSession,
+):
+    """Extract or reuse file content and index it without marking completion.
+
+    The caller marks completion after all its work, including any knowledge linking.
     Note: granular session management is used to prevent connection pool exhaustion.
     The session is committed before external API calls, and updates use a fresh session.
     """
@@ -1714,7 +1732,6 @@ async def process_file(
             hash = calculate_sha256_string(text_content)
 
             if config['rag.bypass_embedding_and_retrieval']:
-                await Files.update_file_data_by_id(file.id, {'status': 'completed', 'error': None}, db=db)
                 await Files.update_file_hash_by_id(file.id, hash, db=db)
                 await publish_event(
                     request,
@@ -1771,11 +1788,6 @@ async def process_file(
                                 db=session,
                             )
 
-                            await Files.update_file_data_by_id(
-                                file.id,
-                                {'status': 'completed', 'error': None},
-                                db=session,
-                            )
                             await Files.update_file_hash_by_id(file.id, hash, db=session)
 
                             await publish_event(