diff --git a/app/tasks/process_document.py b/app/tasks/process_document.py index 0148bdbe..17f3f8e7 100644 --- a/app/tasks/process_document.py +++ b/app/tasks/process_document.py @@ -183,13 +183,6 @@ def process_document( # Not a duplicate (or deduplication disabled) -> insert a new record logger.info(f"[{task_id}] Creating new file record in database") - if settings.enable_deduplication and settings.show_deduplication_step: - log_task_progress( - task_id, - "check_for_duplicates", - "success", - "New file - no duplicates found", - ) log_task_progress(task_id, "create_file_record", "in_progress", "Creating file record") new_record = FileRecord( filehash=filehash, @@ -213,6 +206,17 @@ def process_document( f"File record ID: {new_record.id}", file_id=new_record.id, ) + # Update the check_for_duplicates step now that file_id is available. + # This must happen after initialize_file_steps() which creates the + # step as "pending". The dedup check already passed at this point. + if settings.enable_deduplication: + log_task_progress( + task_id, + "check_for_duplicates", + "success", + "New file - no duplicates found", + file_id=new_record.id, + ) # 1. Generate a UUID-based filename for storage file_ext = os.path.splitext(original_local_file)[1] @@ -330,6 +334,14 @@ def process_document( "Non-PDF file detected, converting to PDF", file_id=file_id, ) + # Mark local text extraction as skipped since the file needs PDF conversion first + log_task_progress( + task_id, + "extract_text", + "skipped", + "Non-PDF file, text extraction deferred to OCR after conversion", + file_id=file_id, + ) log_task_progress( task_id, "process_document", diff --git a/app/utils/file_status.py b/app/utils/file_status.py index a07e5a60..75c84526 100644 --- a/app/utils/file_status.py +++ b/app/utils/file_status.py @@ -145,19 +145,26 @@ def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, D has_errors = failed_steps > 0 # Determine overall status + # + # The pipeline is dynamic: steps may be skipped, added, or + # left "pending" depending on the file type and processing path. + # The terminal step is the authoritative completion signal. + terminal_step = next((s for s in file_steps if s.step_name == TERMINAL_STEP), None) + if has_errors: status = "failed" elif in_progress_steps > 0: status = "processing" elif completed_steps + skipped_steps == total_steps: - # Only mark as completed if the terminal processing step has been - # recorded. Without this guard, files where later pipeline steps - # have not yet started would be falsely marked as "completed". - existing_step_names = {s.step_name for s in file_steps} - if TERMINAL_STEP in existing_step_names: + if terminal_step is not None: status = "completed" else: status = "pending" + elif terminal_step is not None and terminal_step.status == "success": + # Terminal step succeeded but some intermediate steps are + # still "pending" (dynamic pipeline artifacts). The file + # is effectively complete. + status = "completed" else: status = "pending" diff --git a/app/utils/step_manager.py b/app/utils/step_manager.py index 56ad8b58..9e4449a3 100644 --- a/app/utils/step_manager.py +++ b/app/utils/step_manager.py @@ -251,19 +251,32 @@ def get_file_overall_status(db: Session, file_id: int) -> Dict: has_errors = failed_steps > 0 # Determine overall status + # + # The processing pipeline is not strictly linear: steps may be skipped, + # repeated, or dynamically added depending on the file (e.g. OCR is + # skipped when embedded text is found, local extraction is skipped for + # non-PDF files, dedup check may not record its result). Because of + # this, we use the terminal step as the authoritative signal that the + # pipeline finished successfully, rather than requiring every single + # intermediate step to be explicitly marked as success/skipped. + terminal_step_obj = next((s for s in steps if s.step_name == TERMINAL_STEP), None) + if has_errors: status = "failed" elif in_progress_steps > 0: status = "processing" elif completed_steps + skipped_steps == total_steps: - # Only mark as completed if the terminal processing step has been recorded. - # Without this guard, files where later pipeline steps have not yet started - # would be falsely marked as "completed" (e.g. only the first 3 steps ran). - existing_step_names = {s.step_name for s in steps} - if TERMINAL_STEP in existing_step_names: + # All steps resolved – completed only if terminal step was recorded. + if terminal_step_obj is not None: status = "completed" else: status = "pending" + elif terminal_step_obj is not None and terminal_step_obj.status == "success": + # The terminal step succeeded but some intermediate steps are still + # "pending" (e.g. check_for_duplicates logged without file_id, or + # extract_text not marked when OCR path was taken). The pipeline + # is effectively complete. + status = "completed" else: status = "pending" diff --git a/tests/test_file_status_fix.py b/tests/test_file_status_fix.py index 6892eb7e..0a617cf7 100644 --- a/tests/test_file_status_fix.py +++ b/tests/test_file_status_fix.py @@ -474,3 +474,54 @@ class TestFileStatusMissingCoverage: result = get_files_processing_status(db_session, [file_record.id]) assert result[file_record.id]["status"] == "completed" + + def test_get_files_processing_status_completed_with_pending_intermediate(self, db_session): + """Test that terminal step success marks file completed even with pending intermediate steps. + + This tests the dynamic pipeline scenario where check_for_duplicates or + extract_text might remain pending because the pipeline skipped them. + """ + from datetime import datetime + from unittest.mock import patch + + from app.models import FileProcessingStep + from app.utils.file_status import get_files_processing_status + + file_record = FileRecord( + filehash="terminal_fallback", + original_filename="fallback.pdf", + local_filename="/tmp/fallback.pdf", + file_size=100, + ) + db_session.add(file_record) + db_session.commit() + + now = datetime.now() + # Simulate: check_for_duplicates pending, rest succeeded, OCR skipped + for step_name, step_status in [ + ("check_for_duplicates", "pending"), + ("create_file_record", "success"), + ("check_text", "success"), + ("extract_text", "success"), + ("process_with_ocr", "skipped"), + ("extract_metadata_with_gpt", "success"), + ("embed_metadata_into_pdf", "success"), + ("finalize_document_storage", "success"), + ("send_to_all_destinations", "success"), + ]: + step = FileProcessingStep( + file_id=file_record.id, + step_name=step_name, + status=step_status, + created_at=now, + updated_at=now, + ) + db_session.add(step) + db_session.commit() + + with patch("app.config.settings") as ms: + ms.enable_deduplication = True + result = get_files_processing_status(db_session, [file_record.id]) + + # Terminal step succeeded → file should be completed despite pending dedup step + assert result[file_record.id]["status"] == "completed" diff --git a/tests/test_step_manager.py b/tests/test_step_manager.py index 34344b86..2c3cdb47 100644 --- a/tests/test_step_manager.py +++ b/tests/test_step_manager.py @@ -281,6 +281,67 @@ class TestStepManager: assert status["completed_steps"] == len(MAIN_PROCESSING_STEPS) assert status["in_progress_steps"] == 0 + def test_get_file_overall_status_completed_with_pending_intermediate_steps(self, db_session: Session): + """Test that a file is 'completed' when the terminal step succeeds even if some intermediate steps are pending. + + This covers the scenario where steps like check_for_duplicates or + extract_text are left in 'pending' because the dynamic pipeline + skipped them without explicitly marking them (e.g. dedup log written + before file_id was available, or extract_text not marked for non-PDF). + """ + file_record = FileRecord( + filehash="test_terminal_complete", + original_filename="terminal.pdf", + local_filename="/tmp/terminal.pdf", + file_size=1024, + ) + db_session.add(file_record) + db_session.commit() + + initialize_file_steps(db_session, file_record.id) + + # Simulate a pipeline where most steps succeed but check_for_duplicates + # stays pending (the exact scenario from the bug report). + for step_name in MAIN_PROCESSING_STEPS: + if step_name == "check_for_duplicates": + continue # Leave this as "pending" + update_step_status(db_session, file_record.id, step_name, "success") + + # Also add a dynamically created OCR step as "skipped" + update_step_status(db_session, file_record.id, "process_with_ocr", "skipped") + + status = get_file_overall_status(db_session, file_record.id) + + # The terminal step (send_to_all_destinations) succeeded, so the + # file should be "completed" despite check_for_duplicates being pending. + assert status["status"] == "completed" + assert status["has_errors"] is False + + def test_get_file_overall_status_pending_without_terminal_step(self, db_session: Session): + """Test that a file stays 'pending' when the terminal step hasn't been recorded.""" + file_record = FileRecord( + filehash="test_no_terminal", + original_filename="no_terminal.pdf", + local_filename="/tmp/no_terminal.pdf", + file_size=1024, + ) + db_session.add(file_record) + db_session.commit() + + initialize_file_steps(db_session, file_record.id) + + # Only first few steps completed - terminal step is still pending + update_step_status(db_session, file_record.id, "create_file_record", "success") + update_step_status(db_session, file_record.id, "check_text", "success") + update_step_status(db_session, file_record.id, "extract_text", "success") + + status = get_file_overall_status(db_session, file_record.id) + + # Terminal step hasn't run yet, so file should remain pending + assert status["status"] == "pending" + assert status["has_errors"] is False + assert status["completed_steps"] == 3 + def test_get_file_overall_status_failed(self, db_session: Session): """Test overall status for a file with failed steps.""" file_record = FileRecord(