diff --git a/app/models.py b/app/models.py
index 07dc3a7d..9be67422 100644
--- a/app/models.py
+++ b/app/models.py
@@ -1,6 +1,6 @@
 # app/models.py
 
-from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Text, func
+from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, func
 
 from app.database import Base
 
@@ -40,6 +40,27 @@ class FileRecord(Base):
     created_at = Column(DateTime(timezone=True), server_default=func.now())
 
 
+class FileProcessingStep(Base):
+    """
+    Tracks the current status of each processing step for a file.
+    This provides a definitive, queryable state for each step without scanning logs.
+    """
+
+    __tablename__ = "file_processing_steps"
+
+    id = Column(Integer, primary_key=True, index=True)
+    file_id = Column(Integer, ForeignKey("files.id"), nullable=False, index=True)
+    step_name = Column(String, nullable=False, index=True)  # e.g., "hash_file", "upload_to_dropbox"
+    status = Column(String, nullable=False)  # "pending", "in_progress", "success", "failure", "skipped"
+    started_at = Column(DateTime(timezone=True), nullable=True)  # When step started
+    completed_at = Column(DateTime(timezone=True), nullable=True)  # When step finished (success/failure)
+    error_message = Column(Text, nullable=True)  # Error message if status is "failure"
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
+
+    __table_args__ = (UniqueConstraint("file_id", "step_name", name="unique_file_step"),)
+
+
 class ProcessingLog(Base):
     __tablename__ = "processing_logs"
     id = Column(Integer, primary_key=True, index=True)
diff --git a/app/utils/file_status.py b/app/utils/file_status.py
index 0c575926..bb683303 100644
--- a/app/utils/file_status.py
+++ b/app/utils/file_status.py
@@ -6,12 +6,15 @@ from typing import Dict, List
 
 from sqlalchemy.orm import Session
 
-from app.models import ProcessingLog
+from app.models import FileProcessingStep, ProcessingLog
+from app.utils.step_manager import get_file_overall_status, get_step_summary
 
 
 def get_file_processing_status(db: Session, file_id: int) -> Dict:
     """
-    Get the processing status for a file by checking its processing logs.
+    Get the processing status for a file by checking its processing steps.
+
+    This function now queries the FileProcessingStep table instead of scanning logs.
 
     Args:
         db: Database session
@@ -20,12 +23,23 @@ def get_file_processing_status(db: Session, file_id: int) -> Dict:
     Returns:
         dict with status, last_step, and has_errors
     """
-    # Get all logs for this file
-    logs = (
-        db.query(ProcessingLog).filter(ProcessingLog.file_id == file_id).order_by(ProcessingLog.timestamp.desc()).all()
+    # Use the new status table approach
+    overall_status = get_file_overall_status(db, file_id)
+
+    # Get the most recently updated step to determine last_step
+    latest_step = (
+        db.query(FileProcessingStep)
+        .filter(FileProcessingStep.file_id == file_id)
+        .order_by(FileProcessingStep.updated_at.desc())
+        .first()
     )
 
-    return _compute_status_from_logs(logs)
+    return {
+        "status": overall_status["status"],
+        "last_step": latest_step.step_name if latest_step else None,
+        "has_errors": overall_status["has_errors"],
+        "total_steps": overall_status["total_steps"],
+    }
 
 
 def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, Dict]:
@@ -39,26 +53,51 @@ def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, D
     Returns:
         dict mapping file_id to status dict
     """
-    # Get all logs for these files in one query
-    logs = (
-        db.query(ProcessingLog)
-        .filter(ProcessingLog.file_id.in_(file_ids))
-        .order_by(ProcessingLog.file_id, ProcessingLog.timestamp.desc())
-        .all()
-    )
+    # Get all steps for these files in one query
+    steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id.in_(file_ids)).all()
 
-    # Group logs by file_id
-    logs_by_file = {}
-    for log in logs:
-        if log.file_id not in logs_by_file:
-            logs_by_file[log.file_id] = []
-        logs_by_file[log.file_id].append(log)
+    # Group steps by file_id
+    steps_by_file = {}
+    for step in steps:
+        if step.file_id not in steps_by_file:
+            steps_by_file[step.file_id] = []
+        steps_by_file[step.file_id].append(step)
 
     # Compute status for each file
     result = {}
     for file_id in file_ids:
-        file_logs = logs_by_file.get(file_id, [])
-        result[file_id] = _compute_status_from_logs(file_logs)
+        file_steps = steps_by_file.get(file_id, [])
+        if not file_steps:
+            result[file_id] = {"status": "pending", "last_step": None, "has_errors": False, "total_steps": 0}
+        else:
+            # Compute overall status from steps
+            total_steps = len(file_steps)
+            completed_steps = sum(1 for s in file_steps if s.status == "success")
+            failed_steps = sum(1 for s in file_steps if s.status == "failure")
+            in_progress_steps = sum(1 for s in file_steps if s.status == "in_progress")
+            skipped_steps = sum(1 for s in file_steps if s.status == "skipped")
+
+            has_errors = failed_steps > 0
+
+            # Determine overall status
+            if has_errors:
+                status = "failed"
+            elif in_progress_steps > 0:
+                status = "processing"
+            elif completed_steps + skipped_steps == total_steps:
+                status = "completed"
+            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)
+
+            result[file_id] = {
+                "status": status,
+                "last_step": latest_step.step_name,
+                "has_errors": has_errors,
+                "total_steps": total_steps,
+            }
 
     return result
 
@@ -67,6 +106,9 @@ def _compute_status_from_logs(logs: List[ProcessingLog]) -> Dict:
     """
     Compute processing status from a list of processing logs.
 
+    DEPRECATED: This function is kept for backwards compatibility.
+    New code should use the FileProcessingStep table instead.
+
     Args:
         logs: List of ProcessingLog objects (should be ordered by timestamp desc)
 
diff --git a/app/utils/migrate_logs_to_steps.py b/app/utils/migrate_logs_to_steps.py
new file mode 100644
index 00000000..8d148804
--- /dev/null
+++ b/app/utils/migrate_logs_to_steps.py
@@ -0,0 +1,326 @@
+"""
+Migration utility to backfill FileProcessingStep table from existing ProcessingLog entries.
+
+This script analyzes historical logs and populates the FileProcessingStep table
+for files that were processed before the status tracking table was created.
+"""
+
+import logging
+from datetime import datetime
+from typing import Dict, List
+
+from sqlalchemy import func
+from sqlalchemy.orm import Session
+
+from app.models import FileProcessingStep, FileRecord, ProcessingLog
+from app.utils.step_manager import MAIN_PROCESSING_STEPS
+
+logger = logging.getLogger(__name__)
+
+
+def migrate_logs_to_steps(db: Session, file_id: int, dry_run: bool = False) -> Dict:
+    """
+    Migrate ProcessingLog entries to FileProcessingStep for a single file.
+
+    Analyzes all logs for the file and creates/updates FileProcessingStep entries
+    based on the latest status per step found in the logs.
+
+    Args:
+        db: Database session
+        file_id: ID of the file to migrate
+        dry_run: If True, don't commit changes, just report what would be done
+
+    Returns:
+        Dictionary with migration results:
+        {
+            "file_id": 123,
+            "steps_created": 5,
+            "steps_updated": 3,
+            "steps_skipped": 2,
+            "errors": []
+        }
+    """
+    results = {"file_id": file_id, "steps_created": 0, "steps_updated": 0, "steps_skipped": 0, "errors": []}
+
+    try:
+        # Get all logs for this file, ordered by timestamp
+        logs = (
+            db.query(ProcessingLog)
+            .filter(ProcessingLog.file_id == file_id)
+            .order_by(ProcessingLog.timestamp.asc())
+            .all()
+        )
+
+        if not logs:
+            logger.info(f"No logs found for file {file_id}")
+            return results
+
+        # Parse logs to extract latest status per step
+        step_states = _parse_logs_to_step_states(logs)
+
+        # Create or update FileProcessingStep entries
+        for step_name, state in step_states.items():
+            existing_step = (
+                db.query(FileProcessingStep)
+                .filter(FileProcessingStep.file_id == file_id, FileProcessingStep.step_name == step_name)
+                .first()
+            )
+
+            if existing_step:
+                # Update existing step only if it differs
+                if (
+                    existing_step.status != state["status"]
+                    or existing_step.started_at != state["started_at"]
+                    or existing_step.completed_at != state["completed_at"]
+                ):
+                    logger.info(
+                        f"Updating step {step_name} for file {file_id}: {existing_step.status} -> {state['status']}"
+                    )
+                    existing_step.status = state["status"]
+                    existing_step.started_at = state["started_at"]
+                    existing_step.completed_at = state["completed_at"]
+                    existing_step.error_message = state["error_message"]
+                    results["steps_updated"] += 1
+                else:
+                    logger.debug(f"Step {step_name} for file {file_id} already up to date")
+                    results["steps_skipped"] += 1
+            else:
+                # Create new step
+                logger.info(f"Creating step {step_name} for file {file_id} with status {state['status']}")
+                new_step = FileProcessingStep(
+                    file_id=file_id,
+                    step_name=step_name,
+                    status=state["status"],
+                    started_at=state["started_at"],
+                    completed_at=state["completed_at"],
+                    error_message=state["error_message"],
+                )
+                db.add(new_step)
+                results["steps_created"] += 1
+
+        if not dry_run:
+            db.commit()
+            logger.info(
+                f"Migration complete for file {file_id}: "
+                f"{results['steps_created']} created, "
+                f"{results['steps_updated']} updated, "
+                f"{results['steps_skipped']} skipped"
+            )
+        else:
+            db.rollback()
+            logger.info(f"Dry run complete for file {file_id} (no changes committed)")
+
+    except Exception as e:
+        logger.error(f"Error migrating file {file_id}: {str(e)}")
+        results["errors"].append(str(e))
+        db.rollback()
+
+    return results
+
+
+def _parse_logs_to_step_states(logs: List[ProcessingLog]) -> Dict[str, Dict]:
+    """
+    Parse logs to determine the final state of each step.
+
+    Processes logs in chronological order to build a state machine
+    tracking the progression of each step.
+
+    Args:
+        logs: List of ProcessingLog entries ordered by timestamp
+
+    Returns:
+        Dictionary mapping step_name to state dict:
+        {
+            "step_name": {
+                "status": "success",
+                "started_at": datetime,
+                "completed_at": datetime,
+                "error_message": None
+            }
+        }
+    """
+    step_states = {}
+
+    for log in logs:
+        step_name = log.step_name
+        status = log.status.lower()
+
+        # Initialize step state if first time seeing this step
+        if step_name not in step_states:
+            step_states[step_name] = {
+                "status": "pending",
+                "started_at": None,
+                "completed_at": None,
+                "error_message": None,
+            }
+
+        # Update state based on log entry
+        if status == "in_progress":
+            # Step started
+            if step_states[step_name]["started_at"] is None:
+                step_states[step_name]["started_at"] = log.timestamp
+            step_states[step_name]["status"] = "in_progress"
+
+        elif status == "success":
+            # Step completed successfully
+            if step_states[step_name]["started_at"] is None:
+                # If we never saw in_progress, assume it started around this time
+                step_states[step_name]["started_at"] = log.timestamp
+            step_states[step_name]["status"] = "success"
+            step_states[step_name]["completed_at"] = log.timestamp
+            step_states[step_name]["error_message"] = None
+
+        elif status == "failure":
+            # Step failed
+            if step_states[step_name]["started_at"] is None:
+                step_states[step_name]["started_at"] = log.timestamp
+            step_states[step_name]["status"] = "failure"
+            step_states[step_name]["completed_at"] = log.timestamp
+            step_states[step_name]["error_message"] = log.message
+
+        elif status in ["pending", "queued"]:
+            # Only update if step hasn't started yet
+            if step_states[step_name]["status"] == "pending":
+                step_states[step_name]["status"] = "pending"
+
+        # Note: If we see success/failure after a previous success/failure,
+        # the later one wins (represents retry/reprocessing)
+
+    return step_states
+
+
+def migrate_all_files(db: Session, batch_size: int = 100, dry_run: bool = False) -> Dict:
+    """
+    Migrate all files that have logs but no FileProcessingStep entries.
+
+    Args:
+        db: Database session
+        batch_size: Number of files to process in each batch
+        dry_run: If True, don't commit changes
+
+    Returns:
+        Dictionary with overall migration statistics:
+        {
+            "total_files": 150,
+            "files_migrated": 145,
+            "files_failed": 5,
+            "total_steps_created": 1200,
+            "total_steps_updated": 350,
+            "errors": [...]
+        }
+    """
+    summary = {
+        "total_files": 0,
+        "files_migrated": 0,
+        "files_failed": 0,
+        "total_steps_created": 0,
+        "total_steps_updated": 0,
+        "total_steps_skipped": 0,
+        "errors": [],
+    }
+
+    # Find all files that have logs but no steps
+    files_with_logs = db.query(ProcessingLog.file_id).distinct().all()
+    file_ids_with_logs = {file_id for (file_id,) in files_with_logs if file_id is not None}
+
+    files_with_steps = db.query(FileProcessingStep.file_id).distinct().all()
+    file_ids_with_steps = {file_id for (file_id,) in files_with_steps}
+
+    files_to_migrate = list(file_ids_with_logs - file_ids_with_steps)
+
+    summary["total_files"] = len(files_to_migrate)
+    logger.info(f"Found {len(files_to_migrate)} files to migrate")
+
+    # Process in batches
+    for i in range(0, len(files_to_migrate), batch_size):
+        batch = files_to_migrate[i : i + batch_size]
+        logger.info(
+            f"Processing batch {i // batch_size + 1}: files {i + 1} to {min(i + batch_size, len(files_to_migrate))}"
+        )
+
+        for file_id in batch:
+            result = migrate_logs_to_steps(db, file_id, dry_run=dry_run)
+
+            if result["errors"]:
+                summary["files_failed"] += 1
+                summary["errors"].extend(result["errors"])
+            else:
+                summary["files_migrated"] += 1
+                summary["total_steps_created"] += result["steps_created"]
+                summary["total_steps_updated"] += result["steps_updated"]
+                summary["total_steps_skipped"] += result["steps_skipped"]
+
+    logger.info(
+        f"Migration summary: {summary['files_migrated']} files migrated, "
+        f"{summary['files_failed']} failed, "
+        f"{summary['total_steps_created']} steps created, "
+        f"{summary['total_steps_updated']} steps updated"
+    )
+
+    return summary
+
+
+def verify_migration(db: Session, file_id: int) -> Dict:
+    """
+    Verify that migration for a file is correct by comparing logs to steps.
+
+    Args:
+        db: Database session
+        file_id: ID of the file to verify
+
+    Returns:
+        Dictionary with verification results:
+        {
+            "file_id": 123,
+            "is_valid": True,
+            "discrepancies": [],
+            "log_steps": ["step1", "step2"],
+            "table_steps": ["step1", "step2"]
+        }
+    """
+    result = {"file_id": file_id, "is_valid": True, "discrepancies": [], "log_steps": [], "table_steps": []}
+
+    # Get logs and parse them
+    logs = (
+        db.query(ProcessingLog).filter(ProcessingLog.file_id == file_id).order_by(ProcessingLog.timestamp.asc()).all()
+    )
+
+    if not logs:
+        result["discrepancies"].append("No logs found for file")
+        result["is_valid"] = False
+        return result
+
+    expected_states = _parse_logs_to_step_states(logs)
+    result["log_steps"] = sorted(expected_states.keys())
+
+    # Get actual steps from table
+    actual_steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all()
+    result["table_steps"] = sorted([s.step_name for s in actual_steps])
+
+    # Compare
+    actual_states = {s.step_name: s for s in actual_steps}
+
+    # Check for missing steps
+    for step_name in expected_states:
+        if step_name not in actual_states:
+            result["discrepancies"].append(f"Step '{step_name}' missing from table")
+            result["is_valid"] = False
+            continue
+
+        # Compare status
+        expected = expected_states[step_name]
+        actual = actual_states[step_name]
+
+        if expected["status"] != actual.status:
+            result["discrepancies"].append(
+                f"Step '{step_name}' status mismatch: " f"expected '{expected['status']}', got '{actual.status}'"
+            )
+            result["is_valid"] = False
+
+    # Check for extra steps in table
+    for step_name in actual_states:
+        if step_name not in expected_states:
+            result["discrepancies"].append(f"Extra step '{step_name}' in table not found in logs")
+            # Not marking as invalid since this might be intentional
+
+    return result
diff --git a/app/utils/step_manager.py b/app/utils/step_manager.py
new file mode 100644
index 00000000..fc5b58f5
--- /dev/null
+++ b/app/utils/step_manager.py
@@ -0,0 +1,266 @@
+"""
+Utility functions for managing file processing step status.
+
+This module provides functions to initialize, update, and query the status
+of file processing steps using the FileProcessingStep model.
+"""
+
+from datetime import datetime
+from typing import Dict, List, Optional
+
+from sqlalchemy.orm import Session
+
+from app.models import FileProcessingStep
+
+# Define the expected processing steps for a standard file workflow
+MAIN_PROCESSING_STEPS = [
+    "hash_file",
+    "create_file_record",
+    "check_text",
+    "extract_text",  # Or "process_with_azure_document_intelligence"
+    "extract_metadata_with_gpt",
+    "embed_metadata_into_pdf",
+    "finalize_document_storage",
+    "send_to_all_destinations",
+]
+
+
+def initialize_file_steps(db: Session, file_id: int, include_uploads: bool = False) -> None:
+    """
+    Initialize processing steps for a file with 'pending' status.
+
+    Args:
+        db: Database session
+        file_id: ID of the file
+        include_uploads: Whether to include upload destination steps (set after destinations are known)
+    """
+    # Initialize main processing steps
+    for step_name in MAIN_PROCESSING_STEPS:
+        step = FileProcessingStep(file_id=file_id, step_name=step_name, status="pending")
+        db.add(step)
+
+    db.commit()
+
+
+def add_upload_steps(db: Session, file_id: int, upload_destinations: List[str]) -> None:
+    """
+    Add upload destination steps for a file.
+
+    Args:
+        db: Database session
+        file_id: ID of the file
+        upload_destinations: List of upload destination names (e.g., ["dropbox", "s3", "nextcloud"])
+    """
+    for destination in upload_destinations:
+        # Add both queue and upload steps for each destination
+        for prefix in ["queue_", "upload_to_"]:
+            step_name = f"{prefix}{destination}"
+            # Check if step already exists
+            existing = (
+                db.query(FileProcessingStep)
+                .filter(FileProcessingStep.file_id == file_id, FileProcessingStep.step_name == step_name)
+                .first()
+            )
+            if not existing:
+                step = FileProcessingStep(file_id=file_id, step_name=step_name, status="pending")
+                db.add(step)
+
+    db.commit()
+
+
+def update_step_status(
+    db: Session,
+    file_id: int,
+    step_name: str,
+    status: str,
+    error_message: Optional[str] = None,
+    started_at: Optional[datetime] = None,
+    completed_at: Optional[datetime] = None,
+) -> None:
+    """
+    Update the status of a processing step.
+
+    Args:
+        db: Database session
+        file_id: ID of the file
+        step_name: Name of the processing step
+        status: New status ("pending", "in_progress", "success", "failure", "skipped")
+        error_message: Error message if status is "failure"
+        started_at: When the step started (for "in_progress")
+        completed_at: When the step completed (for "success" or "failure")
+    """
+    step = (
+        db.query(FileProcessingStep)
+        .filter(FileProcessingStep.file_id == file_id, FileProcessingStep.step_name == step_name)
+        .first()
+    )
+
+    if not step:
+        # Create the step if it doesn't exist
+        step = FileProcessingStep(
+            file_id=file_id,
+            step_name=step_name,
+            status=status,
+            started_at=started_at,
+            completed_at=completed_at,
+            error_message=error_message,
+        )
+        db.add(step)
+    else:
+        # Update existing step
+        step.status = status
+        if error_message is not None:
+            step.error_message = error_message
+        if started_at is not None:
+            step.started_at = started_at
+        if completed_at is not None:
+            step.completed_at = completed_at
+
+    db.commit()
+
+
+def get_file_step_status(db: Session, file_id: int) -> Dict[str, Dict]:
+    """
+    Get the current status of all processing steps for a file.
+
+    Args:
+        db: Database session
+        file_id: ID of the file
+
+    Returns:
+        Dictionary mapping step_name to status info:
+        {
+            "step_name": {
+                "status": "success",
+                "started_at": datetime,
+                "completed_at": datetime,
+                "error_message": None
+            },
+            ...
+        }
+    """
+    steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all()
+
+    result = {}
+    for step in steps:
+        result[step.step_name] = {
+            "status": step.status,
+            "started_at": step.started_at,
+            "completed_at": step.completed_at,
+            "error_message": step.error_message,
+        }
+
+    return result
+
+
+def get_file_overall_status(db: Session, file_id: int) -> Dict:
+    """
+    Get the overall processing status for a file based on its steps.
+
+    Args:
+        db: Database session
+        file_id: ID of the file
+
+    Returns:
+        Dictionary with overall status info:
+        {
+            "status": "completed",  # "pending", "processing", "completed", "failed"
+            "has_errors": False,
+            "total_steps": 10,
+            "completed_steps": 8,
+            "failed_steps": 0,
+            "in_progress_steps": 2
+        }
+    """
+    steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all()
+
+    if not steps:
+        return {
+            "status": "pending",
+            "has_errors": False,
+            "total_steps": 0,
+            "completed_steps": 0,
+            "failed_steps": 0,
+            "in_progress_steps": 0,
+        }
+
+    total_steps = len(steps)
+    completed_steps = sum(1 for s in steps if s.status == "success")
+    failed_steps = sum(1 for s in steps if s.status == "failure")
+    in_progress_steps = sum(1 for s in steps if s.status == "in_progress")
+    skipped_steps = sum(1 for s in steps if s.status == "skipped")
+
+    has_errors = failed_steps > 0
+
+    # Determine overall status
+    if has_errors:
+        status = "failed"
+    elif in_progress_steps > 0:
+        status = "processing"
+    elif completed_steps + skipped_steps == total_steps:
+        status = "completed"
+    else:
+        status = "pending"
+
+    return {
+        "status": status,
+        "has_errors": has_errors,
+        "total_steps": total_steps,
+        "completed_steps": completed_steps,
+        "failed_steps": failed_steps,
+        "in_progress_steps": in_progress_steps,
+        "skipped_steps": skipped_steps,
+    }
+
+
+def get_step_summary(db: Session, file_id: int) -> Dict:
+    """
+    Get a summary of main steps vs upload steps with status counts.
+
+    Args:
+        db: Database session
+        file_id: ID of the file
+
+    Returns:
+        Dictionary with step counts:
+        {
+            "main": {"queued": 0, "in_progress": 1, "success": 7, "failure": 0, "skipped": 0},
+            "uploads": {"queued": 2, "in_progress": 0, "success": 4, "failure": 0, "skipped": 0},
+            "total_main_steps": 8,
+            "total_upload_tasks": 6
+        }
+    """
+    steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all()
+
+    main_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0, "skipped": 0}
+    upload_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0, "skipped": 0}
+
+    upload_prefixes = ["upload_to_", "queue_"]
+
+    main_steps_count = 0
+    upload_steps_count = 0
+
+    for step in steps:
+        # Normalize status
+        status = step.status.lower()
+        if status == "pending":
+            status = "queued"
+
+        # Check if it's an upload task
+        is_upload = any(step.step_name.startswith(prefix) for prefix in upload_prefixes)
+
+        if is_upload:
+            if status in upload_counts:
+                upload_counts[status] += 1
+            upload_steps_count += 1
+        elif step.step_name in MAIN_PROCESSING_STEPS:
+            if status in main_counts:
+                main_counts[status] += 1
+            main_steps_count += 1
+
+    return {
+        "main": main_counts,
+        "uploads": upload_counts,
+        "total_main_steps": main_steps_count,
+        "total_upload_tasks": upload_steps_count,
+    }
diff --git a/app/views/files.py b/app/views/files.py
index 545178f1..f259c566 100644
--- a/app/views/files.py
+++ b/app/views/files.py
@@ -175,8 +175,14 @@ def file_detail_page(request: Request, file_id: int, db: Session = Depends(get_d
         # Compute processing flow for visualization
         flow_data = _compute_processing_flow(logs)
 
-        # Compute step-aligned summary
-        step_summary = _compute_step_summary(logs)
+        # Compute step-aligned summary from status table (preferred) or fallback to logs
+        try:
+            from app.utils.step_manager import get_step_summary as get_step_summary_from_table
+
+            step_summary = get_step_summary_from_table(db, file_id)
+        except Exception:
+            # Fallback to log-based computation if status table not available
+            step_summary = _compute_step_summary(logs)
 
         return templates.TemplateResponse(
             "file_detail.html",
@@ -328,6 +334,9 @@ def _compute_step_summary(logs):
     Compute a step-aligned summary from logs showing queued, success, and failure counts.
 
     Returns a dictionary with main step counts and upload branch counts.
+
+    Note: This function is order-independent - it selects the latest status per step
+    based on timestamp, regardless of input log ordering.
     """
     # Count statuses for main processing steps (not uploads)
     main_steps = [
@@ -347,9 +356,9 @@ def _compute_step_summary(logs):
     main_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0}
     upload_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0}
 
-    # Track latest status for each step (logs are ordered by timestamp desc)
-    main_steps_seen = {}
-    upload_tasks_seen = {}
+    # Track latest status for each step by comparing timestamps (order-independent)
+    main_steps_seen = {}  # {step_name: (timestamp, status)}
+    upload_tasks_seen = {}  # {step_name: (timestamp, status)}
 
     for log in logs:
         step_name = log.step_name
@@ -363,21 +372,21 @@ def _compute_step_summary(logs):
         is_upload = any(step_name.startswith(prefix) for prefix in upload_prefixes)
 
         if is_upload:
-            # Track latest status for each unique upload task (first seen is latest)
-            if step_name not in upload_tasks_seen:
-                upload_tasks_seen[step_name] = status
+            # Track latest status for each unique upload task by timestamp
+            if step_name not in upload_tasks_seen or log.timestamp > upload_tasks_seen[step_name][0]:
+                upload_tasks_seen[step_name] = (log.timestamp, status)
         elif step_name in main_steps:
-            # Track latest status for main steps (first seen is latest)
-            if step_name not in main_steps_seen:
-                main_steps_seen[step_name] = status
+            # Track latest status for main steps by timestamp
+            if step_name not in main_steps_seen or log.timestamp > main_steps_seen[step_name][0]:
+                main_steps_seen[step_name] = (log.timestamp, status)
 
     # Count main step statuses from latest status per step
-    for task_status in main_steps_seen.values():
+    for _, task_status in main_steps_seen.values():
         if task_status in main_counts:
             main_counts[task_status] += 1
 
     # Count upload task statuses from latest status per task
-    for task_status in upload_tasks_seen.values():
+    for _, task_status in upload_tasks_seen.values():
         if task_status in upload_counts:
             upload_counts[task_status] += 1
 
diff --git a/docs/FileProcessingStatusArchitecture.md b/docs/FileProcessingStatusArchitecture.md
new file mode 100644
index 00000000..109e98b8
--- /dev/null
+++ b/docs/FileProcessingStatusArchitecture.md
@@ -0,0 +1,354 @@
+# File Processing Status Architecture
+
+## Overview
+
+DocuElevate now uses a **dual-table architecture** for tracking file processing:
+
+1. **`FileProcessingStep`** - Current state tracking (NEW)
+2. **`ProcessingLog`** - Historical audit trail (EXISTING, PRESERVED)
+
+This separation provides both **fast status queries** and **complete audit history**.
+
+## Architecture
+
+### FileProcessingStep Table (Current State)
+
+**Purpose**: Definitive source of truth for current processing status
+
+**Structure**:
+```sql
+CREATE TABLE file_processing_steps (
+    id INTEGER PRIMARY KEY,
+    file_id INTEGER NOT NULL,
+    step_name VARCHAR NOT NULL,
+    status VARCHAR NOT NULL,  -- 'pending', 'in_progress', 'success', 'failure', 'skipped'
+    started_at TIMESTAMP,
+    completed_at TIMESTAMP,
+    error_message TEXT,
+    created_at TIMESTAMP DEFAULT NOW(),
+    updated_at TIMESTAMP DEFAULT NOW(),
+    UNIQUE (file_id, step_name)
+)
+```
+
+**Key Features**:
+- One row per step per file
+- Updated when step status changes
+- No scanning required - direct lookups
+- Indexed for fast queries
+
+### ProcessingLog Table (Audit Trail)
+
+**Purpose**: Complete historical record of all processing events
+
+**Structure**: (Unchanged from original)
+```sql
+CREATE TABLE processing_logs (
+    id INTEGER PRIMARY KEY,
+    file_id INTEGER,
+    task_id VARCHAR,
+    step_name VARCHAR,
+    status VARCHAR,
+    message VARCHAR,
+    detail TEXT,
+    timestamp TIMESTAMP DEFAULT NOW()
+)
+```
+
+**Key Features**:
+- Multiple rows per step (one per event)
+- Never deleted - complete history
+- Used for debugging, compliance, audit
+- Detailed error messages and diagnostics
+
+## How They Work Together
+
+### When a File is Created
+
+```python
+from app.utils.step_manager import initialize_file_steps
+
+# Create file record
+file_record = FileRecord(...)
+db.add(file_record)
+db.commit()
+
+# Initialize expected processing steps
+initialize_file_steps(db, file_record.id)
+# Creates FileProcessingStep rows with status='pending'
+```
+
+### When a Worker Processes a Step
+
+```python
+from app.utils.step_manager import update_step_status
+
+# Worker starts processing
+update_step_status(
+    db, file_id, "hash_file", "in_progress",
+    started_at=datetime.now()
+)
+
+# Worker also logs to ProcessingLog (for history)
+log = ProcessingLog(
+    file_id=file_id,
+    step_name="hash_file",
+    status="in_progress",
+    message="Starting file hash calculation"
+)
+db.add(log)
+db.commit()
+
+# Worker completes successfully
+update_step_status(
+    db, file_id, "hash_file", "success",
+    completed_at=datetime.now()
+)
+
+# Also log completion
+log = ProcessingLog(
+    file_id=file_id,
+    step_name="hash_file",
+    status="success",
+    message="Hash calculated successfully: abc123..."
+)
+db.add(log)
+db.commit()
+```
+
+### When Dashboard Queries Status
+
+```python
+from app.utils.step_manager import get_file_overall_status, get_step_summary
+
+# Fast lookup - no log scanning!
+overall = get_file_overall_status(db, file_id)
+# Returns: {"status": "processing", "has_errors": False, ...}
+
+# Get detailed step breakdown
+summary = get_step_summary(db, file_id)
+# Returns: {
+#   "main": {"success": 3, "in_progress": 1, "pending": 4},
+#   "uploads": {"success": 2, "queued": 4}
+# }
+```
+
+### When Viewing Processing History
+
+```python
+# Logs provide complete timeline
+logs = db.query(ProcessingLog)\
+    .filter(ProcessingLog.file_id == file_id)\
+    .order_by(ProcessingLog.timestamp.asc())\
+    .all()
+
+# Shows every event with timestamps, messages, details
+# Example:
+# 2024-01-15 10:00:00 | hash_file | in_progress | Starting...
+# 2024-01-15 10:00:05 | hash_file | success | Hash: abc123
+# 2024-01-15 10:00:06 | check_text | in_progress | Checking...
+# ... etc
+```
+
+## Migration from Logs
+
+For existing files with only ProcessingLog entries:
+
+```python
+from app.utils.migrate_logs_to_steps import migrate_logs_to_steps, migrate_all_files
+
+# Migrate single file
+result = migrate_logs_to_steps(db, file_id)
+# Parses logs chronologically to reconstruct final state
+
+# Migrate all files (batch processing)
+summary = migrate_all_files(db, batch_size=100)
+# Finds files with logs but no steps, migrates them all
+
+# Dry run mode (test without committing)
+result = migrate_logs_to_steps(db, file_id, dry_run=True)
+```
+
+The migration utility:
+1. Reads all logs for a file in chronological order
+2. Tracks state transitions (pending → in_progress → success/failure)
+3. Determines final status per step
+4. Creates FileProcessingStep entries
+5. Preserves all original logs unchanged
+
+## Benefits
+
+### Performance
+- **Old**: Scan all logs, dedupe, find latest → O(n log n)
+- **New**: Single table lookup → O(1)
+
+### Correctness
+- **Old**: Dependent on log ordering, can miss retries
+- **New**: Definitive state, always up-to-date
+
+### Flexibility
+- **Old**: Hard-coded step list, can't add dynamic steps
+- **New**: Dynamic upload destinations, custom workflows
+
+### Debugging
+- **Old**: Only logs available
+- **New**: Current state + complete history
+
+## Example Workflow
+
+```python
+# 1. File uploaded
+file = FileRecord(filehash="abc123", ...)
+db.add(file)
+db.commit()
+
+# 2. Initialize steps
+initialize_file_steps(db, file.id)
+# FileProcessingStep: 9 rows with status='pending'
+
+# 3. Add upload destinations (discovered later)
+add_upload_steps(db, file.id, ["dropbox", "s3", "nextcloud"])
+# FileProcessingStep: +6 rows with status='pending'
+
+# 4. Worker processes each step
+for step in ["hash_file", "create_file_record", "check_text", ...]:
+    # Start
+    update_step_status(db, file.id, step, "in_progress", started_at=now())
+    log_event(db, file.id, step, "in_progress", "Starting...")
+    
+    # Do work...
+    
+    # Complete
+    update_step_status(db, file.id, step, "success", completed_at=now())
+    log_event(db, file.id, step, "success", "Completed successfully")
+
+# 5. Dashboard queries
+status = get_file_overall_status(db, file.id)
+# status="completed", all steps done
+
+# 6. User views history
+logs = get_processing_logs(db, file.id)
+# Complete timeline with all events, messages, timestamps
+```
+
+## Status Flow
+
+```
+pending → in_progress → success
+                     ↓
+                   failure
+```
+
+## Main Processing Steps
+
+Defined in `app/utils/step_manager.py`:
+```python
+MAIN_PROCESSING_STEPS = [
+    "hash_file",
+    "create_file_record",
+    "check_text",
+    "extract_text",
+    "extract_metadata_with_gpt",
+    "embed_metadata_into_pdf",
+    "finalize_document_storage",
+    "send_to_all_destinations",
+]
+```
+
+## Upload Steps
+
+Dynamically created per destination:
+- `queue_{destination}` - Queued for upload
+- `upload_to_{destination}` - Actual upload
+
+Example: Dropbox
+- `queue_dropbox` - Added to upload queue
+- `upload_to_dropbox` - Upload to Dropbox
+
+## API Reference
+
+### Step Manager Functions
+
+```python
+# Initialize steps for new file
+initialize_file_steps(db: Session, file_id: int) -> None
+
+# Add upload destination steps
+add_upload_steps(db: Session, file_id: int, destinations: List[str]) -> None
+
+# Update step status
+update_step_status(
+    db: Session,
+    file_id: int,
+    step_name: str,
+    status: str,
+    error_message: Optional[str] = None,
+    started_at: Optional[datetime] = None,
+    completed_at: Optional[datetime] = None
+) -> None
+
+# Query step status
+get_file_step_status(db: Session, file_id: int) -> Dict[str, Dict]
+get_file_overall_status(db: Session, file_id: int) -> Dict
+get_step_summary(db: Session, file_id: int) -> Dict
+```
+
+### Migration Functions
+
+```python
+# Migrate single file
+migrate_logs_to_steps(
+    db: Session,
+    file_id: int,
+    dry_run: bool = False
+) -> Dict
+
+# Migrate all files
+migrate_all_files(
+    db: Session,
+    batch_size: int = 100,
+    dry_run: bool = False
+) -> Dict
+
+# Verify migration
+verify_migration(db: Session, file_id: int) -> Dict
+```
+
+## Best Practices
+
+1. **Always log AND update status**: Workers should do both
+2. **Logs are append-only**: Never delete or modify logs
+3. **Status is current state**: FileProcessingStep reflects latest status
+4. **Initialize early**: Create steps when file is created
+5. **Add uploads dynamically**: Add upload steps when destinations are known
+6. **Handle retries**: update_step_status() handles overwriting previous status
+7. **Preserve history**: Use logs for debugging and compliance
+
+## Troubleshooting
+
+### Status shows wrong state
+- Check FileProcessingStep table directly
+- Use `get_file_step_status()` to see all steps
+- Verify workers are calling `update_step_status()`
+
+### Missing steps
+- Check if `initialize_file_steps()` was called
+- For uploads, check if `add_upload_steps()` was called
+- Use migration utility to backfill from logs
+
+### Dashboard shows old data
+- Clear any caching
+- Verify status table is being queried, not logs
+- Check `app/views/files.py` is using `get_step_summary_from_table()`
+
+## Migration Checklist
+
+For deploying to existing system:
+
+- [ ] Apply database migration: `alembic upgrade head`
+- [ ] Run migration utility: `migrate_all_files(db, dry_run=True)` to test
+- [ ] Run actual migration: `migrate_all_files(db, dry_run=False)`
+- [ ] Verify: Check a few files with `verify_migration()`
+- [ ] Update workers to call `update_step_status()` 
+- [ ] Update file creation to call `initialize_file_steps()`
+- [ ] Monitor dashboard for correct status display
diff --git a/migrations/versions/001_file_processing_steps.py b/migrations/versions/001_file_processing_steps.py
new file mode 100644
index 00000000..8b91930b
--- /dev/null
+++ b/migrations/versions/001_file_processing_steps.py
@@ -0,0 +1,45 @@
+"""Add file processing steps table
+
+Revision ID: 001_file_processing_steps
+Revises:
+Create Date: 2026-02-11
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "001_file_processing_steps"
+down_revision: Union[str, None] = None
+depends_on: Union[str, None] = None
+
+
+def upgrade() -> None:
+    """Create file_processing_steps table."""
+    op.create_table(
+        "file_processing_steps",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("file_id", sa.Integer(), nullable=False),
+        sa.Column("step_name", sa.String(), nullable=False),
+        sa.Column("status", sa.String(), nullable=False),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("error_message", sa.Text(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["file_id"], ["files.id"], name="fk_file_processing_steps_file_id"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("file_id", "step_name", name="unique_file_step"),
+    )
+    op.create_index("ix_file_processing_steps_file_id", "file_processing_steps", ["file_id"])
+    op.create_index("ix_file_processing_steps_step_name", "file_processing_steps", ["step_name"])
+
+
+def downgrade() -> None:
+    """Drop file_processing_steps table."""
+    op.drop_index("ix_file_processing_steps_step_name", table_name="file_processing_steps")
+    op.drop_index("ix_file_processing_steps_file_id", table_name="file_processing_steps")
+    op.drop_table("file_processing_steps")
diff --git a/tests/test_file_detail_endpoints.py b/tests/test_file_detail_endpoints.py
index 6f897ee8..9335453a 100644
--- a/tests/test_file_detail_endpoints.py
+++ b/tests/test_file_detail_endpoints.py
@@ -231,9 +231,7 @@ class TestSubtaskRetry:
 
         with patch("app.tasks.extract_metadata_with_gpt.extract_metadata_with_gpt") as mock_extract:
             mock_extract.delay.return_value = mock_task
-            response = client.post(
-                f"/api/files/{file_record.id}/retry-subtask?subtask_name=embed_metadata_into_pdf"
-            )
+            response = client.post(f"/api/files/{file_record.id}/retry-subtask?subtask_name=embed_metadata_into_pdf")
         assert response.status_code == 200
         data = response.json()
         assert data["status"] == "success"
@@ -503,21 +501,25 @@ class TestStepSummary:
 
     def test_summary_with_mixed_statuses(self):
         """Test step summary with various statuses."""
+        from datetime import datetime, timedelta
+
         from app.views.files import _compute_step_summary
 
         # Create mock logs
         class MockLog:
-            def __init__(self, step_name, status):
+            def __init__(self, step_name, status, timestamp):
                 self.step_name = step_name
                 self.status = status
+                self.timestamp = timestamp
 
+        now = datetime.now()
         logs = [
-            MockLog("hash_file", "success"),
-            MockLog("create_file_record", "success"),
-            MockLog("extract_metadata_with_gpt", "failure"),
-            MockLog("upload_to_dropbox", "success"),
-            MockLog("upload_to_s3", "failure"),
-            MockLog("upload_to_nextcloud", "in_progress"),
+            MockLog("hash_file", "success", now - timedelta(minutes=5)),
+            MockLog("create_file_record", "success", now - timedelta(minutes=4)),
+            MockLog("extract_metadata_with_gpt", "failure", now - timedelta(minutes=3)),
+            MockLog("upload_to_dropbox", "success", now - timedelta(minutes=2)),
+            MockLog("upload_to_s3", "failure", now - timedelta(minutes=1)),
+            MockLog("upload_to_nextcloud", "in_progress", now),
         ]
 
         summary = _compute_step_summary(logs)
diff --git a/tests/test_file_status_fix.py b/tests/test_file_status_fix.py
index 66ba46f3..b44ed09b 100644
--- a/tests/test_file_status_fix.py
+++ b/tests/test_file_status_fix.py
@@ -7,9 +7,10 @@ This test module verifies that:
 3. Files with completed steps show "completed" not "processing"
 """
 
-import pytest
 from datetime import datetime, timedelta
 
+import pytest
+
 from app.utils.file_status import _compute_status_from_logs
 from app.views.files import _compute_step_summary
 
@@ -249,3 +250,82 @@ class TestMetricsCountingBugFixes:
         assert summary["uploads"]["success"] == 4  # 1 upload + 3 queue
         assert summary["uploads"]["failure"] == 1  # 1 upload
         assert summary["uploads"]["in_progress"] == 1  # 1 upload
+
+    def test_order_independent_ascending(self):
+        """
+        Test that _compute_step_summary works correctly with ascending order logs
+        (as used in production by file_detail_page).
+
+        This test ensures the function correctly selects the latest status per step
+        based on timestamp, not position in the list.
+        """
+
+        class MockLog:
+            def __init__(self, step_name, status, timestamp):
+                self.step_name = step_name
+                self.status = status
+                self.timestamp = timestamp
+
+        now = datetime.now()
+        # Logs ordered by timestamp ASCENDING (oldest first) - like production
+        logs = [
+            # Older in_progress logs (should be ignored)
+            MockLog("hash_file", "in_progress", now - timedelta(minutes=7)),
+            MockLog("create_file_record", "in_progress", now - timedelta(minutes=6)),
+            MockLog("extract_metadata_with_gpt", "in_progress", now - timedelta(minutes=5)),
+            MockLog("upload_to_dropbox", "in_progress", now - timedelta(minutes=4)),
+            # Latest status for each step (all success) - at the end
+            MockLog("hash_file", "success", now - timedelta(minutes=3)),
+            MockLog("create_file_record", "success", now - timedelta(minutes=2)),
+            MockLog("extract_metadata_with_gpt", "success", now - timedelta(minutes=1)),
+            MockLog("upload_to_dropbox", "success", now),
+        ]
+
+        summary = _compute_step_summary(logs)
+
+        # Should use latest status (success) not first seen (in_progress)
+        assert summary["total_main_steps"] == 3
+        assert summary["main"]["success"] == 3
+        assert summary["main"]["in_progress"] == 0
+
+        assert summary["total_upload_tasks"] == 1
+        assert summary["uploads"]["success"] == 1
+        assert summary["uploads"]["in_progress"] == 0
+
+    def test_order_independent_mixed(self):
+        """
+        Test that _compute_step_summary works correctly with randomly ordered logs.
+
+        This ensures the function truly is order-independent.
+        """
+
+        class MockLog:
+            def __init__(self, step_name, status, timestamp):
+                self.step_name = step_name
+                self.status = status
+                self.timestamp = timestamp
+
+        now = datetime.now()
+        # Logs in mixed order
+        logs = [
+            MockLog("upload_to_s3", "in_progress", now - timedelta(minutes=8)),
+            MockLog("hash_file", "success", now - timedelta(minutes=1)),  # Latest for hash_file
+            MockLog("upload_to_dropbox", "in_progress", now - timedelta(minutes=7)),
+            MockLog("hash_file", "in_progress", now - timedelta(minutes=5)),  # Older, should be ignored
+            MockLog("upload_to_s3", "failure", now - timedelta(minutes=2)),  # Latest for S3
+            MockLog("create_file_record", "in_progress", now - timedelta(minutes=6)),
+            MockLog("upload_to_dropbox", "success", now),  # Latest for Dropbox
+            MockLog("create_file_record", "success", now - timedelta(minutes=3)),  # Latest for create
+        ]
+
+        summary = _compute_step_summary(logs)
+
+        # Should correctly identify latest status for each step
+        assert summary["total_main_steps"] == 2
+        assert summary["main"]["success"] == 2  # hash_file and create_file_record
+        assert summary["main"]["in_progress"] == 0
+
+        assert summary["total_upload_tasks"] == 2
+        assert summary["uploads"]["success"] == 1  # dropbox
+        assert summary["uploads"]["failure"] == 1  # s3
+        assert summary["uploads"]["in_progress"] == 0
diff --git a/tests/test_step_manager.py b/tests/test_step_manager.py
new file mode 100644
index 00000000..37c13305
--- /dev/null
+++ b/tests/test_step_manager.py
@@ -0,0 +1,331 @@
+"""
+Tests for the FileProcessingStep model and step_manager utilities.
+
+This test module verifies the new explicit status tracking approach.
+"""
+
+from datetime import datetime, timedelta
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from app.database import Base
+from app.models import FileProcessingStep, FileRecord
+from app.utils.step_manager import (
+    MAIN_PROCESSING_STEPS,
+    add_upload_steps,
+    get_file_overall_status,
+    get_file_step_status,
+    get_step_summary,
+    initialize_file_steps,
+    update_step_status,
+)
+
+
+@pytest.fixture
+def db_session():
+    """Create an in-memory SQLite database for testing."""
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    SessionLocal = sessionmaker(bind=engine)
+    session = SessionLocal()
+
+    yield session
+
+    session.close()
+    Base.metadata.drop_all(engine)
+
+
+@pytest.mark.unit
+class TestFileProcessingStepModel:
+    """Test the FileProcessingStep model."""
+
+    def test_create_step(self, db_session: Session):
+        """Test creating a processing step."""
+        # Create a file record first
+        file_record = FileRecord(
+            filehash="test123", original_filename="test.pdf", local_filename="/tmp/test.pdf", file_size=1024
+        )
+        db_session.add(file_record)
+        db_session.commit()
+
+        # Create a processing step
+        step = FileProcessingStep(file_id=file_record.id, step_name="hash_file", status="success")
+        db_session.add(step)
+        db_session.commit()
+
+        assert step.id is not None
+        assert step.file_id == file_record.id
+        assert step.step_name == "hash_file"
+        assert step.status == "success"
+
+    def test_unique_constraint(self, db_session: Session):
+        """Test that the unique constraint on (file_id, step_name) works."""
+        # Create a file record
+        file_record = FileRecord(
+            filehash="test456", original_filename="test2.pdf", local_filename="/tmp/test2.pdf", file_size=2048
+        )
+        db_session.add(file_record)
+        db_session.commit()
+
+        # Create first step
+        step1 = FileProcessingStep(file_id=file_record.id, step_name="hash_file", status="in_progress")
+        db_session.add(step1)
+        db_session.commit()
+
+        # Try to create duplicate step - should fail
+        step2 = FileProcessingStep(file_id=file_record.id, step_name="hash_file", status="success")
+        db_session.add(step2)
+
+        with pytest.raises(Exception):  # SQLAlchemy will raise an integrity error
+            db_session.commit()
+
+
+@pytest.mark.unit
+class TestStepManager:
+    """Test step_manager utility functions."""
+
+    def test_initialize_file_steps(self, db_session: Session):
+        """Test initializing processing steps for a file."""
+        # Create a file record
+        file_record = FileRecord(
+            filehash="test789", original_filename="test3.pdf", local_filename="/tmp/test3.pdf", file_size=4096
+        )
+        db_session.add(file_record)
+        db_session.commit()
+
+        # Initialize steps
+        initialize_file_steps(db_session, file_record.id)
+
+        # Verify steps were created
+        steps = db_session.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_record.id).all()
+
+        assert len(steps) == len(MAIN_PROCESSING_STEPS)
+        for step in steps:
+            assert step.status == "pending"
+            assert step.step_name in MAIN_PROCESSING_STEPS
+
+    def test_add_upload_steps(self, db_session: Session):
+        """Test adding upload destination steps."""
+        # Create a file record
+        file_record = FileRecord(
+            filehash="test101112", original_filename="test4.pdf", local_filename="/tmp/test4.pdf", file_size=8192
+        )
+        db_session.add(file_record)
+        db_session.commit()
+
+        # Add upload steps
+        destinations = ["dropbox", "s3", "nextcloud"]
+        add_upload_steps(db_session, file_record.id, destinations)
+
+        # Verify upload steps were created
+        steps = db_session.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_record.id).all()
+
+        # Should have 2 steps per destination (queue_ and upload_to_)
+        assert len(steps) == len(destinations) * 2
+
+        expected_steps = []
+        for dest in destinations:
+            expected_steps.append(f"queue_{dest}")
+            expected_steps.append(f"upload_to_{dest}")
+
+        for step in steps:
+            assert step.step_name in expected_steps
+            assert step.status == "pending"
+
+    def test_update_step_status_new_step(self, db_session: Session):
+        """Test updating status creates step if it doesn't exist."""
+        # Create a file record
+        file_record = FileRecord(
+            filehash="test131415", original_filename="test5.pdf", local_filename="/tmp/test5.pdf", file_size=16384
+        )
+        db_session.add(file_record)
+        db_session.commit()
+
+        # Update a step that doesn't exist yet
+        now = datetime.now()
+        update_step_status(db_session, file_record.id, "hash_file", "in_progress", started_at=now)
+
+        # Verify step was created
+        step = (
+            db_session.query(FileProcessingStep)
+            .filter(FileProcessingStep.file_id == file_record.id, FileProcessingStep.step_name == "hash_file")
+            .first()
+        )
+
+        assert step is not None
+        assert step.status == "in_progress"
+        assert step.started_at == now
+
+    def test_update_step_status_existing_step(self, db_session: Session):
+        """Test updating status of existing step."""
+        # Create a file record and initialize steps
+        file_record = FileRecord(
+            filehash="test161718", original_filename="test6.pdf", local_filename="/tmp/test6.pdf", file_size=32768
+        )
+        db_session.add(file_record)
+        db_session.commit()
+
+        initialize_file_steps(db_session, file_record.id)
+
+        # Update an existing step
+        now = datetime.now()
+        update_step_status(
+            db_session, file_record.id, "hash_file", "success", started_at=now - timedelta(seconds=5), completed_at=now
+        )
+
+        # Verify step was updated
+        step = (
+            db_session.query(FileProcessingStep)
+            .filter(FileProcessingStep.file_id == file_record.id, FileProcessingStep.step_name == "hash_file")
+            .first()
+        )
+
+        assert step.status == "success"
+        assert step.started_at == now - timedelta(seconds=5)
+        assert step.completed_at == now
+
+    def test_get_file_step_status(self, db_session: Session):
+        """Test retrieving all step statuses for a file."""
+        # Create a file record and initialize steps
+        file_record = FileRecord(
+            filehash="test192021", original_filename="test7.pdf", local_filename="/tmp/test7.pdf", file_size=65536
+        )
+        db_session.add(file_record)
+        db_session.commit()
+
+        initialize_file_steps(db_session, file_record.id)
+
+        # Update some steps
+        now = datetime.now()
+        update_step_status(db_session, file_record.id, "hash_file", "success", completed_at=now)
+        update_step_status(db_session, file_record.id, "create_file_record", "in_progress", started_at=now)
+        update_step_status(db_session, file_record.id, "check_text", "failure", error_message="Failed to check text")
+
+        # Get all step statuses
+        status_map = get_file_step_status(db_session, file_record.id)
+
+        assert len(status_map) == len(MAIN_PROCESSING_STEPS)
+        assert status_map["hash_file"]["status"] == "success"
+        assert status_map["hash_file"]["completed_at"] == now
+        assert status_map["create_file_record"]["status"] == "in_progress"
+        assert status_map["check_text"]["status"] == "failure"
+        assert status_map["check_text"]["error_message"] == "Failed to check text"
+
+    def test_get_file_overall_status_pending(self, db_session: Session):
+        """Test overall status for a file with pending steps."""
+        file_record = FileRecord(
+            filehash="test222324", original_filename="test8.pdf", local_filename="/tmp/test8.pdf", file_size=131072
+        )
+        db_session.add(file_record)
+        db_session.commit()
+
+        initialize_file_steps(db_session, file_record.id)
+
+        status = get_file_overall_status(db_session, file_record.id)
+
+        assert status["status"] == "pending"
+        assert status["has_errors"] is False
+        assert status["total_steps"] == len(MAIN_PROCESSING_STEPS)
+        assert status["completed_steps"] == 0
+        assert status["in_progress_steps"] == 0
+
+    def test_get_file_overall_status_processing(self, db_session: Session):
+        """Test overall status for a file with in-progress steps."""
+        file_record = FileRecord(
+            filehash="test252627", original_filename="test9.pdf", local_filename="/tmp/test9.pdf", file_size=262144
+        )
+        db_session.add(file_record)
+        db_session.commit()
+
+        initialize_file_steps(db_session, file_record.id)
+
+        # Mark some steps as complete and one as in progress
+        update_step_status(db_session, file_record.id, "hash_file", "success")
+        update_step_status(db_session, file_record.id, "create_file_record", "success")
+        update_step_status(db_session, file_record.id, "check_text", "in_progress")
+
+        status = get_file_overall_status(db_session, file_record.id)
+
+        assert status["status"] == "processing"
+        assert status["has_errors"] is False
+        assert status["completed_steps"] == 2
+        assert status["in_progress_steps"] == 1
+
+    def test_get_file_overall_status_completed(self, db_session: Session):
+        """Test overall status for a completed file."""
+        file_record = FileRecord(
+            filehash="test282930", original_filename="test10.pdf", local_filename="/tmp/test10.pdf", file_size=524288
+        )
+        db_session.add(file_record)
+        db_session.commit()
+
+        initialize_file_steps(db_session, file_record.id)
+
+        # Mark all steps as success
+        for step_name in MAIN_PROCESSING_STEPS:
+            update_step_status(db_session, file_record.id, step_name, "success")
+
+        status = get_file_overall_status(db_session, file_record.id)
+
+        assert status["status"] == "completed"
+        assert status["has_errors"] is False
+        assert status["completed_steps"] == len(MAIN_PROCESSING_STEPS)
+        assert status["in_progress_steps"] == 0
+
+    def test_get_file_overall_status_failed(self, db_session: Session):
+        """Test overall status for a file with failed steps."""
+        file_record = FileRecord(
+            filehash="test313233", original_filename="test11.pdf", local_filename="/tmp/test11.pdf", file_size=1048576
+        )
+        db_session.add(file_record)
+        db_session.commit()
+
+        initialize_file_steps(db_session, file_record.id)
+
+        # Mark some steps as success and one as failure
+        update_step_status(db_session, file_record.id, "hash_file", "success")
+        update_step_status(db_session, file_record.id, "create_file_record", "success")
+        update_step_status(db_session, file_record.id, "check_text", "failure", error_message="OCR failed")
+
+        status = get_file_overall_status(db_session, file_record.id)
+
+        assert status["status"] == "failed"
+        assert status["has_errors"] is True
+        assert status["failed_steps"] == 1
+
+    def test_get_step_summary(self, db_session: Session):
+        """Test getting step summary with counts."""
+        file_record = FileRecord(
+            filehash="test343536", original_filename="test12.pdf", local_filename="/tmp/test12.pdf", file_size=2097152
+        )
+        db_session.add(file_record)
+        db_session.commit()
+
+        # Initialize main steps and add upload steps
+        initialize_file_steps(db_session, file_record.id)
+        add_upload_steps(db_session, file_record.id, ["dropbox", "s3", "nextcloud"])
+
+        # Update statuses
+        update_step_status(db_session, file_record.id, "hash_file", "success")
+        update_step_status(db_session, file_record.id, "create_file_record", "success")
+        update_step_status(db_session, file_record.id, "check_text", "in_progress")
+        update_step_status(db_session, file_record.id, "upload_to_dropbox", "success")
+        update_step_status(db_session, file_record.id, "upload_to_s3", "failure")
+        update_step_status(db_session, file_record.id, "queue_nextcloud", "in_progress")
+
+        summary = get_step_summary(db_session, file_record.id)
+
+        # Check main step counts
+        assert summary["main"]["success"] == 2
+        assert summary["main"]["in_progress"] == 1
+        assert summary["main"]["queued"] >= 5  # Remaining pending steps
+
+        # Check upload counts
+        assert summary["uploads"]["success"] == 1
+        assert summary["uploads"]["failure"] == 1
+        assert summary["uploads"]["in_progress"] == 1
+
+        assert summary["total_main_steps"] == len(MAIN_PROCESSING_STEPS)
+        assert summary["total_upload_tasks"] == 6  # 3 destinations x 2 steps each