From e6dd39c27d4b8e77a9c546c911eace2783b47068 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 27 Feb 2026 00:51:14 +0000 Subject: [PATCH] fix(status): prevent false Completed status when mandatory pipeline steps have not run Add a terminal-step guard (send_to_all_destinations) to all status calculation paths so that files are only marked Completed once the entire processing pipeline has been recorded. - get_file_overall_status: require TERMINAL_STEP to be present - get_files_processing_status: same guard for bulk status - get_step_summary: count missing terminal step as queued so total_main_steps > main_completed when pipeline is incomplete - apply_status_filter: SQL sub-query requires terminal step for completed filter - process_document: call initialize_file_steps after creating a new file record so all mandatory steps are pre-created as pending Define TERMINAL_STEP constant in step_manager.py and reference it in file_status.py and file_queries.py to avoid magic strings. Tests updated: add send_to_all_destinations to completed-file fixtures; add test verifying initialize_file_steps is called for new files. Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/tasks/process_document.py | 4 ++ app/utils/file_queries.py | 20 +++++-- app/utils/file_status.py | 11 +++- app/utils/step_manager.py | 24 ++++++++- tests/test_file_queries.py | 16 ++++-- tests/test_file_status_fix.py | 3 +- tests/test_process_document.py | 99 ++++++++++++++++++++++++++++++++++ 7 files changed, 165 insertions(+), 12 deletions(-) diff --git a/app/tasks/process_document.py b/app/tasks/process_document.py index 5167485e..0148bdbe 100644 --- a/app/tasks/process_document.py +++ b/app/tasks/process_document.py @@ -17,6 +17,7 @@ from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt from app.tasks.process_with_ocr import process_with_ocr from app.tasks.retry_config import BaseTaskWithRetry from app.utils import get_unique_filepath_with_counter, hash_file, log_task_progress +from app.utils.step_manager import initialize_file_steps from app.utils.text_quality import check_text_quality, detect_pdf_text_source logger = logging.getLogger(__name__) @@ -202,6 +203,9 @@ def process_document( db.commit() db.refresh(new_record) logger.info(f"[{task_id}] File record created with ID: {new_record.id}") + # Pre-initialize all expected processing steps as "pending" so that + # status tracking reflects the complete pipeline from the start. + initialize_file_steps(db, new_record.id) log_task_progress( task_id, "create_file_record", diff --git a/app/utils/file_queries.py b/app/utils/file_queries.py index 4fa79e9f..7e7c86af 100644 --- a/app/utils/file_queries.py +++ b/app/utils/file_queries.py @@ -11,6 +11,7 @@ from sqlalchemy import or_ from sqlalchemy.orm import Query, Session from app.models import FileProcessingStep, FileRecord +from app.utils.step_manager import TERMINAL_STEP def apply_status_filter(query: Query, db: Session, status: Optional[str]) -> Query: @@ -98,6 +99,9 @@ def apply_status_filter(query: Query, db: Session, status: Optional[str]) -> Que query = query.filter(FileRecord.id.in_(db.query(subq.c.file_id))) elif status == "completed": # Files where all real steps are either success or skipped (no failures or in_progress) + # and the terminal send_to_all_destinations step has been recorded. + # Excluding the terminal-step requirement allows files that only completed the + # first few pipeline stages to be falsely labelled as "completed". # Exclude duplicates from completed query = query.filter(FileRecord.is_duplicate.is_(False)) @@ -113,9 +117,19 @@ def apply_status_filter(query: Query, db: Session, status: Optional[str]) -> Que .subquery() ) - # Select files with real steps that don't have issues - query = query.filter(FileRecord.id.in_(db.query(files_with_real_steps.c.file_id))).filter( - ~FileRecord.id.in_(db.query(files_with_issues.c.file_id)) + # Get files that have the terminal processing step recorded + files_with_terminal_step = ( + db.query(FileProcessingStep.file_id) + .filter(FileProcessingStep.step_name == TERMINAL_STEP) + .distinct() + .subquery() + ) + + # Select files with real steps, no issues, and terminal step present + query = ( + query.filter(FileRecord.id.in_(db.query(files_with_real_steps.c.file_id))) + .filter(~FileRecord.id.in_(db.query(files_with_issues.c.file_id))) + .filter(FileRecord.id.in_(db.query(files_with_terminal_step.c.file_id))) ) elif status == "duplicate": # Files marked as duplicates diff --git a/app/utils/file_status.py b/app/utils/file_status.py index 5a278a7c..a07e5a60 100644 --- a/app/utils/file_status.py +++ b/app/utils/file_status.py @@ -7,7 +7,7 @@ from typing import Dict, List from sqlalchemy.orm import Session from app.models import FileProcessingStep, FileRecord, ProcessingLog -from app.utils.step_manager import get_file_overall_status +from app.utils.step_manager import TERMINAL_STEP, get_file_overall_status def get_file_processing_status(db: Session, file_id: int) -> Dict: @@ -150,7 +150,14 @@ def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, D elif in_progress_steps > 0: status = "processing" elif completed_steps + skipped_steps == total_steps: - status = "completed" + # 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: + status = "completed" + else: + status = "pending" else: status = "pending" diff --git a/app/utils/step_manager.py b/app/utils/step_manager.py index 0d84ee09..d183452e 100644 --- a/app/utils/step_manager.py +++ b/app/utils/step_manager.py @@ -29,6 +29,10 @@ OPTIONAL_PROCESSING_STEPS = { "check_for_duplicates": settings.enable_deduplication, # Only if deduplication is enabled } +# The terminal step is the last mandatory step in the processing pipeline. +# A file is only considered "completed" once this step has been recorded. +TERMINAL_STEP = "send_to_all_destinations" + # Combine steps based on configuration MAIN_PROCESSING_STEPS = [] if settings.enable_deduplication: @@ -252,7 +256,14 @@ def get_file_overall_status(db: Session, file_id: int) -> Dict: elif in_progress_steps > 0: status = "processing" elif completed_steps + skipped_steps == total_steps: - status = "completed" + # 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: + status = "completed" + else: + status = "pending" else: status = "pending" @@ -340,6 +351,17 @@ def get_step_summary(db: Session, file_id: int) -> Dict: main_counts[status] += 1 main_steps_count += 1 + # Ensure the terminal step is always counted in total_main_steps. + # If the terminal step has not been recorded yet, the pipeline is not + # complete; counting it as "queued" prevents the status banner from + # showing "Completed" before the full pipeline has run. + # Note: if TERMINAL_STEP already appears in `steps`, the loop above has + # already incremented `main_steps_count` for it, so we only add here when + # the step is absent from the DB entirely. + if not any(step.step_name == TERMINAL_STEP for step in steps if step.step_name in REAL_MAIN_STEPS): + main_steps_count += 1 + main_counts["queued"] += 1 + return { "main": main_counts, "uploads": upload_counts, diff --git a/tests/test_file_queries.py b/tests/test_file_queries.py index 1352a2ab..6e20f335 100644 --- a/tests/test_file_queries.py +++ b/tests/test_file_queries.py @@ -76,7 +76,7 @@ def sample_files(db_session): ) db_session.add(step3) - # File 4: completed (has success step, no failures) + # File 4: completed (has success step including terminal step) file4 = FileRecord( filehash="hash4", original_filename="completed.pdf", @@ -87,10 +87,11 @@ def sample_files(db_session): db_session.add(file4) db_session.flush() - step4 = FileProcessingStep(file_id=file4.id, step_name="extract_text", status="success") - db_session.add(step4) + step4a = FileProcessingStep(file_id=file4.id, step_name="extract_text", status="success") + step4b = FileProcessingStep(file_id=file4.id, step_name="send_to_all_destinations", status="success") + db_session.add_all([step4a, step4b]) - # File 5: completed with multiple success steps + # File 5: completed with multiple success steps including terminal step file5 = FileRecord( filehash="hash5", original_filename="completed2.pdf", @@ -111,7 +112,12 @@ def sample_files(db_session): step_name="extract_metadata_with_gpt", status="success", ) - db_session.add_all([step5a, step5b]) + step5c = FileProcessingStep( + file_id=file5.id, + step_name="send_to_all_destinations", + status="success", + ) + db_session.add_all([step5a, step5b, step5c]) # File 6: has success but also failure (should be filtered out from completed) file6 = FileRecord( diff --git a/tests/test_file_status_fix.py b/tests/test_file_status_fix.py index 8d3eff60..6892eb7e 100644 --- a/tests/test_file_status_fix.py +++ b/tests/test_file_status_fix.py @@ -437,7 +437,7 @@ class TestFileStatusMissingCoverage: assert result["has_errors"] is False def test_get_files_processing_status_with_completed_steps(self, db_session): - """Covers line 189->188: completed + skipped == total_steps → completed status.""" + """Covers completed + skipped == total_steps with terminal step → completed status.""" from datetime import datetime from unittest.mock import patch @@ -457,6 +457,7 @@ class TestFileStatusMissingCoverage: for step_name, step_status in [ ("create_file_record", "success"), ("finalize_document_storage", "skipped"), + ("send_to_all_destinations", "success"), ]: step = FileProcessingStep( file_id=file_record.id, diff --git a/tests/test_process_document.py b/tests/test_process_document.py index ce4d2149..fbabdec9 100644 --- a/tests/test_process_document.py +++ b/tests/test_process_document.py @@ -743,3 +743,102 @@ startxref # Verify that retry was triggered # The retry method raises a special exception assert exc_info.value is not None + + +@pytest.mark.unit +@pytest.mark.requires_db +def test_process_document_initializes_file_steps_for_new_file(db_session, tmp_path): + """ + Test that process_document calls initialize_file_steps for new file records + so that all mandatory pipeline steps are pre-created as "pending". + + This ensures status tracking reflects the complete expected pipeline from + the start and prevents incomplete files from being falsely marked as + "completed" just because the steps that *did* run all succeeded. + """ + # Create a test PDF file with embedded text + test_pdf = tmp_path / "test.pdf" + pdf_content = b"""%PDF-1.4 +1 0 obj +<< +/Type /Catalog +/Pages 2 0 R +>> +endobj +2 0 obj +<< +/Type /Pages +/Kids [3 0 R] +/Count 1 +>> +endobj +3 0 obj +<< +/Type /Page +/Parent 2 0 R +/MediaBox [0 0 612 792] +/Resources << +/Font << +/F1 << +/Type /Font +/Subtype /Type1 +/BaseFont /Helvetica +>> +>> +>> +/Contents 4 0 R +>> +endobj +4 0 obj +<< +/Length 44 +>> +stream +BT +/F1 12 Tf +100 700 Td +(Test content) Tj +ET +endstream +endobj +xref +0 5 +0000000000 65535 f +0000000009 00000 n +0000000058 00000 n +0000000115 00000 n +0000000306 00000 n +trailer +<< +/Size 5 +/Root 1 0 R +>> +startxref +399 +%%EOF +""" + test_pdf.write_bytes(pdf_content) + + with ( + patch("app.tasks.process_document.SessionLocal") as mock_session_local, + patch("app.tasks.process_document.settings") as mock_settings, + patch("app.tasks.process_document.log_task_progress"), + patch("app.tasks.process_document.extract_metadata_with_gpt") as mock_extract, + patch("app.tasks.process_document.initialize_file_steps") as mock_init_steps, + ): + mock_settings.workdir = str(tmp_path) + mock_settings.enable_deduplication = False + mock_settings.enable_text_quality_check = False + mock_session_local.return_value.__enter__.return_value = db_session + mock_session_local.return_value.__exit__.return_value = None + mock_extract.delay = MagicMock() + + result = process_document.run(str(test_pdf)) + + assert result["status"] == "Text extracted locally" + assert "file_id" in result + + # initialize_file_steps must have been called exactly once with the new file's ID + mock_init_steps.assert_called_once() + called_file_id = mock_init_steps.call_args[0][1] + assert called_file_id == result["file_id"]