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

@@ -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",

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
+        # Get files that have the terminal processing step recorded
-        query = query.filter(FileRecord.id.in_(db.query(files_with_real_steps.c.file_id))).filter(
+        files_with_terminal_step = (
-            ~FileRecord.id.in_(db.query(files_with_issues.c.file_id))
+            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

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,9 +150,16 @@ 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:
+                # 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"
 
             # Get last updated step
             latest_step = max(file_steps, key=lambda s: s.updated_at if s.updated_at else s.created_at)

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,9 +256,16 @@ 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:
+        # 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"
 
     return {
         "status": status,
@@ -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,

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")
+    step4a = FileProcessingStep(file_id=file4.id, step_name="extract_text", status="success")
-    db_session.add(step4)
+    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(

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,

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"]