diff --git a/app/api/files.py b/app/api/files.py
index d055782e..b99521af 100644
--- a/app/api/files.py
+++ b/app/api/files.py
@@ -312,8 +312,8 @@ def bulk_reprocess_files(request: Request, file_ids: List[int], db: Session = De
)
continue
- # Queue the file for processing
- task = process_document.delay(file_record.local_filename)
+ # Queue the file for processing, passing file_id to skip duplicate check
+ task = process_document.delay(file_record.local_filename, file_id=file_record.id)
task_ids.append(task.id)
processed_files.append(
{"file_id": file_record.id, "filename": file_record.original_filename, "task_id": task.id}
@@ -366,8 +366,10 @@ def reprocess_single_file(request: Request, file_id: int, db: Session = Depends(
if not file_record.local_filename or not os.path.exists(file_record.local_filename):
raise HTTPException(status_code=400, detail="Local file not found on disk. Cannot reprocess.")
- # Queue the file for processing
- task = process_document.delay(file_record.local_filename, original_filename=file_record.original_filename)
+ # Queue the file for processing, passing file_id to skip duplicate check
+ task = process_document.delay(
+ file_record.local_filename, original_filename=file_record.original_filename, file_id=file_record.id
+ )
logger.info(
f"Reprocessing file: ID={file_record.id}, " f"Filename={file_record.original_filename}, TaskID={task.id}"
@@ -388,20 +390,130 @@ def reprocess_single_file(request: Request, file_id: int, db: Session = Depends(
raise HTTPException(status_code=500, detail=f"Error reprocessing file: {str(e)}")
+def _retry_pipeline_step(file_record: FileRecord, step_name: str, db: Session) -> dict:
+ """
+ Retry a specific pipeline processing step for a file.
+
+ Supports restarting from intermediate pipeline steps:
+ - process_document: Full reprocessing (skips duplicate check)
+ - process_with_azure_document_intelligence: OCR processing
+ - extract_metadata_with_gpt: Metadata extraction
+ - embed_metadata_into_pdf: Metadata embedding
+
+ Args:
+ file_record: The FileRecord to reprocess
+ step_name: Name of the pipeline step to retry
+ db: Database session
+
+ Returns:
+ Dict with task ID and status information
+ """
+ file_id = file_record.id
+
+ if step_name == "process_document":
+ # Full reprocessing with duplicate check bypass
+ if not file_record.local_filename or not os.path.exists(file_record.local_filename):
+ raise HTTPException(status_code=400, detail="Local file not found on disk. Cannot retry.")
+ task = process_document.delay(
+ file_record.local_filename, original_filename=file_record.original_filename, file_id=file_id
+ )
+ elif step_name == "process_with_azure_document_intelligence":
+ from app.tasks.process_with_azure_document_intelligence import process_with_azure_document_intelligence
+
+ # OCR needs the file in workdir/tmp
+ if not file_record.local_filename or not os.path.exists(file_record.local_filename):
+ raise HTTPException(status_code=400, detail="Local file not found on disk. Cannot retry OCR.")
+ filename = os.path.basename(file_record.local_filename)
+ task = process_with_azure_document_intelligence.delay(filename, file_id)
+ elif step_name == "extract_metadata_with_gpt":
+ from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt
+
+ # Extract text from the file to pass to GPT
+ if not file_record.local_filename or not os.path.exists(file_record.local_filename):
+ raise HTTPException(
+ status_code=400, detail="Local file not found on disk. Cannot retry metadata extraction."
+ )
+ import PyPDF2
+
+ extracted_text = ""
+ with open(file_record.local_filename, "rb") as f:
+ pdf_reader = PyPDF2.PdfReader(f)
+ for page in pdf_reader.pages:
+ extracted_text += page.extract_text() + "\n"
+
+ filename = os.path.basename(file_record.local_filename)
+ task = extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
+ elif step_name == "embed_metadata_into_pdf":
+ from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf
+
+ # Retrieve the last successful metadata extraction result from processing logs
+ last_metadata_log = (
+ db.query(ProcessingLog)
+ .filter(
+ ProcessingLog.file_id == file_id,
+ ProcessingLog.step_name == "extract_metadata_with_gpt",
+ ProcessingLog.status == "success",
+ )
+ .order_by(ProcessingLog.timestamp.desc())
+ .first()
+ )
+ if not last_metadata_log:
+ raise HTTPException(
+ status_code=400,
+ detail="No successful metadata extraction found. Retry extract_metadata_with_gpt first.",
+ )
+
+ if not file_record.local_filename or not os.path.exists(file_record.local_filename):
+ raise HTTPException(
+ status_code=400, detail="Local file not found on disk. Cannot retry metadata embedding."
+ )
+ # Re-extract text and metadata for embedding
+ import PyPDF2
+
+ extracted_text = ""
+ with open(file_record.local_filename, "rb") as f:
+ pdf_reader = PyPDF2.PdfReader(f)
+ for page in pdf_reader.pages:
+ extracted_text += page.extract_text() + "\n"
+
+ filename = os.path.basename(file_record.local_filename)
+ # Pass empty metadata dict - the embed task will use whatever was last extracted
+ # The actual metadata should ideally be stored, but for retry we re-extract
+ task = embed_metadata_into_pdf.delay(filename, extracted_text, {}, file_id)
+ else:
+ raise HTTPException(status_code=400, detail=f"Unsupported pipeline step: {step_name}")
+
+ logger.info(f"Retrying pipeline step: FileID={file_record.id}, Step={step_name}, TaskID={task.id}")
+
+ return {
+ "status": "success",
+ "message": f"Pipeline step {step_name} queued for retry",
+ "file_id": file_record.id,
+ "subtask_name": step_name,
+ "task_id": task.id,
+ }
+
+
@router.post("/files/{file_id}/retry-subtask")
@require_login
def retry_subtask(
request: Request,
file_id: int,
- subtask_name: str = Query(..., description="Name of the upload subtask to retry (e.g., 'upload_to_dropbox')"),
+ subtask_name: str = Query(
+ ..., description="Name of the subtask to retry (e.g., 'upload_to_dropbox', 'extract_metadata_with_gpt')"
+ ),
db: Session = Depends(get_db),
):
"""
- Retry a specific failed upload subtask for a file.
+ Retry a specific failed subtask for a file.
+
+ Supports both upload tasks (e.g., upload_to_dropbox) and pipeline processing
+ steps (e.g., process_with_azure_document_intelligence, extract_metadata_with_gpt,
+ embed_metadata_into_pdf).
Args:
file_id: ID of the file
- subtask_name: Name of the upload task (e.g., upload_to_dropbox, upload_to_s3)
+ subtask_name: Name of the task to retry
Returns:
Task ID and status information
@@ -413,6 +525,17 @@ def retry_subtask(
if not file_record:
raise HTTPException(status_code=404, detail=f"File with ID {file_id} not found")
+ # Pipeline processing steps that can be retried from the failed step
+ pipeline_step_names = {
+ "process_document",
+ "process_with_azure_document_intelligence",
+ "extract_metadata_with_gpt",
+ "embed_metadata_into_pdf",
+ }
+
+ if subtask_name in pipeline_step_names:
+ return _retry_pipeline_step(file_record, subtask_name, db)
+
# Map subtask names to their corresponding Celery tasks
from app.tasks.upload_to_dropbox import upload_to_dropbox
from app.tasks.upload_to_email import upload_to_email
@@ -439,9 +562,10 @@ def retry_subtask(
}
if subtask_name not in task_map:
+ all_valid = sorted(list(task_map.keys()) + sorted(pipeline_step_names))
raise HTTPException(
status_code=400,
- detail=f"Invalid subtask name: {subtask_name}. Must be one of: {', '.join(task_map.keys())}",
+ detail=f"Invalid subtask name: {subtask_name}. Must be one of: {', '.join(all_valid)}",
)
# Check for processed file (upload tasks work with processed files)
diff --git a/app/tasks/process_document.py b/app/tasks/process_document.py
index c43de473..55cff021 100644
--- a/app/tasks/process_document.py
+++ b/app/tasks/process_document.py
@@ -23,16 +23,19 @@ logger = logging.getLogger(__name__)
@celery.task(base=BaseTaskWithRetry, bind=True)
-def process_document(self, original_local_file: str, original_filename: str = None):
+def process_document(self, original_local_file: str, original_filename: str = None, file_id: int = None):
"""
Process a document file and trigger appropriate text extraction.
Args:
original_local_file: Path to the file on disk
original_filename: Optional original filename (if different from path basename)
+ file_id: Optional existing file record ID. When provided, skips duplicate
+ detection and reuses the existing record (used for reprocessing).
Steps:
1. Check if we have a FileRecord entry (via SHA-256 hash). If found, skip re-processing.
+ (Skipped when file_id is provided for reprocessing.)
2. If not found, insert a new DB row and continue with the pipeline:
- Copy file to /workdir/tmp
- Check for embedded text. If present, run local GPT extraction
@@ -74,43 +77,61 @@ def process_document(self, original_local_file: str, original_filename: str = No
# Acquire DB session in the task
with SessionLocal() as db:
- existing = db.query(FileRecord).filter_by(filehash=filehash).one_or_none()
- if existing:
- logger.info(f"[{task_id}] Duplicate file detected (hash={filehash[:10]}...) Skipping processing.")
+ # When file_id is provided, we are reprocessing an existing file.
+ # Skip the duplicate check and reuse the existing record.
+ if file_id is not None:
+ existing_record = db.query(FileRecord).filter_by(id=file_id).one_or_none()
+ if existing_record is None:
+ logger.error(f"[{task_id}] File record with ID {file_id} not found for reprocessing.")
+ log_task_progress(task_id, "process_document", "failure", "File record not found", file_id=file_id)
+ return {"error": "File record not found", "file_id": file_id}
+ logger.info(f"[{task_id}] Reprocessing existing file record ID: {file_id}, skipping duplicate check.")
log_task_progress(
task_id,
"process_document",
- "success",
- "Duplicate file detected, skipping",
- file_id=existing.id,
+ "in_progress",
+ f"Reprocessing file record ID: {file_id}",
+ file_id=file_id,
)
- return {
- "status": "duplicate_file",
- "file_id": existing.id,
- "detail": "File already processed.",
- }
+ new_record = existing_record
+ else:
+ existing = db.query(FileRecord).filter_by(filehash=filehash).one_or_none()
+ if existing:
+ logger.info(f"[{task_id}] Duplicate file detected (hash={filehash[:10]}...) Skipping processing.")
+ log_task_progress(
+ task_id,
+ "process_document",
+ "success",
+ "Duplicate file detected, skipping",
+ file_id=existing.id,
+ )
+ return {
+ "status": "duplicate_file",
+ "file_id": existing.id,
+ "detail": "File already processed.",
+ }
- # Not a duplicate -> insert a new record
- logger.info(f"[{task_id}] Creating new file record in database")
- log_task_progress(task_id, "create_file_record", "in_progress", "Creating file record")
- new_record = FileRecord(
- filehash=filehash,
- original_filename=original_filename,
- local_filename="", # Will fill in after we move it
- file_size=file_size,
- mime_type=mime_type,
- )
- db.add(new_record)
- db.commit()
- db.refresh(new_record)
- logger.info(f"[{task_id}] File record created with ID: {new_record.id}")
- log_task_progress(
- task_id,
- "create_file_record",
- "success",
- f"File record ID: {new_record.id}",
- file_id=new_record.id,
- )
+ # Not a duplicate -> insert a new record
+ logger.info(f"[{task_id}] Creating new file record in database")
+ log_task_progress(task_id, "create_file_record", "in_progress", "Creating file record")
+ new_record = FileRecord(
+ filehash=filehash,
+ original_filename=original_filename,
+ local_filename="", # Will fill in after we move it
+ file_size=file_size,
+ mime_type=mime_type,
+ )
+ db.add(new_record)
+ db.commit()
+ db.refresh(new_record)
+ logger.info(f"[{task_id}] File record created with ID: {new_record.id}")
+ log_task_progress(
+ task_id,
+ "create_file_record",
+ "success",
+ f"File record ID: {new_record.id}",
+ file_id=new_record.id,
+ )
# 1. Generate a UUID-based filename and place it in /workdir/tmp
file_ext = os.path.splitext(original_local_file)[1]
diff --git a/frontend/templates/file_detail.html b/frontend/templates/file_detail.html
index 35203cbf..675ed0d9 100644
--- a/frontend/templates/file_detail.html
+++ b/frontend/templates/file_detail.html
@@ -799,6 +799,15 @@
{% if stage.status == 'not_run' %}
Not executed
{% endif %}
+ {% if stage.can_retry and stage.status == 'failure' %}
+
+ {% endif %}
{% if stage.timestamp %}
{{ stage.timestamp.strftime('%Y-%m-%d %H:%M:%S') }}
diff --git a/tests/test_bulk_operations.py b/tests/test_bulk_operations.py
index adfd1d21..79ac00d5 100644
--- a/tests/test_bulk_operations.py
+++ b/tests/test_bulk_operations.py
@@ -2,10 +2,12 @@
Tests for bulk file operations (delete and reprocess).
"""
+from unittest.mock import MagicMock, patch
+
import pytest
from fastapi.testclient import TestClient
+
from app.models import FileRecord, ProcessingLog
-from unittest.mock import patch, MagicMock
@pytest.mark.integration
@@ -94,7 +96,7 @@ class TestBulkOperations:
@patch("app.api.files.process_document")
def test_bulk_reprocess_success(self, mock_process_document, client: TestClient, db_session):
- """Test bulk reprocessing of files."""
+ """Test bulk reprocessing of files passes file_id to skip duplicate check."""
# Setup mock
mock_task = MagicMock()
mock_task.id = "test-task-id"
@@ -125,6 +127,10 @@ class TestBulkOperations:
assert len(data["processed_files"]) == 2
assert len(data["task_ids"]) == 2
+ # Verify that file_id was passed to skip duplicate check
+ for call_args in mock_process_document.delay.call_args_list:
+ assert "file_id" in call_args.kwargs or len(call_args.args) > 1
+
@patch("app.api.files.process_document")
def test_bulk_reprocess_missing_files(self, mock_process_document, client: TestClient, db_session):
"""Test bulk reprocessing when some local files are missing."""
diff --git a/tests/test_file_detail_endpoints.py b/tests/test_file_detail_endpoints.py
index d7b52233..dc1aac75 100644
--- a/tests/test_file_detail_endpoints.py
+++ b/tests/test_file_detail_endpoints.py
@@ -3,9 +3,11 @@ Tests for file detail view improvements including reprocessing and preview endpo
"""
import os
+from unittest.mock import MagicMock, patch
+
import pytest
-from unittest.mock import patch, MagicMock
from fastapi.testclient import TestClient
+
from app.models import FileRecord, ProcessingLog
@@ -127,6 +129,127 @@ class TestSubtaskRetry:
assert response.status_code == 400
assert "processed file not found" in response.json()["detail"].lower()
+ @patch("app.api.files.process_document")
+ def test_retry_pipeline_step_process_document(
+ self, mock_process_document, client: TestClient, db_session, sample_pdf_path
+ ):
+ """Test retrying the process_document pipeline step."""
+ mock_task = MagicMock()
+ mock_task.id = "retry-task-123"
+ mock_process_document.delay.return_value = mock_task
+
+ file_record = FileRecord(
+ filehash="pipeline_retry1",
+ original_filename="pipeline.pdf",
+ local_filename=sample_pdf_path,
+ file_size=1024,
+ mime_type="application/pdf",
+ )
+ db_session.add(file_record)
+ db_session.commit()
+ db_session.refresh(file_record)
+
+ response = client.post(f"/api/files/{file_record.id}/retry-subtask?subtask_name=process_document")
+ assert response.status_code == 200
+ data = response.json()
+ assert data["status"] == "success"
+ assert data["subtask_name"] == "process_document"
+ assert "task_id" in data
+
+ # Verify process_document.delay was called with file_id
+ mock_process_document.delay.assert_called_once()
+ call_kwargs = mock_process_document.delay.call_args
+ assert call_kwargs[1].get("file_id") == file_record.id or call_kwargs[0][-1] == file_record.id
+
+ def test_retry_pipeline_step_ocr(self, client: TestClient, db_session, sample_pdf_path):
+ """Test retrying the OCR pipeline step."""
+ mock_task = MagicMock()
+ mock_task.id = "ocr-retry-task"
+
+ file_record = FileRecord(
+ filehash="pipeline_retry2",
+ original_filename="ocr_retry.pdf",
+ local_filename=sample_pdf_path,
+ file_size=1024,
+ mime_type="application/pdf",
+ )
+ db_session.add(file_record)
+ db_session.commit()
+ db_session.refresh(file_record)
+
+ with patch(
+ "app.tasks.process_with_azure_document_intelligence.process_with_azure_document_intelligence"
+ ) as mock_azure:
+ mock_azure.delay.return_value = mock_task
+ response = client.post(
+ f"/api/files/{file_record.id}/retry-subtask?subtask_name=process_with_azure_document_intelligence"
+ )
+ assert response.status_code == 200
+ data = response.json()
+ assert data["status"] == "success"
+ assert data["subtask_name"] == "process_with_azure_document_intelligence"
+
+ def test_retry_pipeline_step_metadata_extraction(self, client: TestClient, db_session, sample_pdf_path):
+ """Test retrying the metadata extraction pipeline step."""
+ mock_task = MagicMock()
+ mock_task.id = "gpt-retry-task"
+
+ file_record = FileRecord(
+ filehash="pipeline_retry3",
+ original_filename="gpt_retry.pdf",
+ local_filename=sample_pdf_path,
+ file_size=1024,
+ mime_type="application/pdf",
+ )
+ db_session.add(file_record)
+ db_session.commit()
+ db_session.refresh(file_record)
+
+ with patch("app.tasks.extract_metadata_with_gpt.extract_metadata_with_gpt") as mock_gpt:
+ mock_gpt.delay.return_value = mock_task
+ response = client.post(f"/api/files/{file_record.id}/retry-subtask?subtask_name=extract_metadata_with_gpt")
+ assert response.status_code == 200
+ data = response.json()
+ assert data["status"] == "success"
+ assert data["subtask_name"] == "extract_metadata_with_gpt"
+
+ def test_retry_pipeline_step_embed_no_metadata(self, client: TestClient, db_session, sample_pdf_path):
+ """Test retrying embed_metadata_into_pdf without prior metadata extraction."""
+ file_record = FileRecord(
+ filehash="pipeline_retry4",
+ original_filename="embed_retry.pdf",
+ local_filename=sample_pdf_path,
+ file_size=1024,
+ mime_type="application/pdf",
+ )
+ db_session.add(file_record)
+ db_session.commit()
+ db_session.refresh(file_record)
+
+ # Try to retry embed without a successful metadata extraction log
+ response = client.post(f"/api/files/{file_record.id}/retry-subtask?subtask_name=embed_metadata_into_pdf")
+ assert response.status_code == 400
+ assert "retry extract_metadata_with_gpt first" in response.json()["detail"].lower()
+
+ def test_retry_pipeline_step_missing_local_file(self, client: TestClient, db_session):
+ """Test retrying a pipeline step when local file is missing."""
+ file_record = FileRecord(
+ filehash="pipeline_retry5",
+ original_filename="missing.pdf",
+ local_filename="/nonexistent/path/missing.pdf",
+ file_size=1024,
+ mime_type="application/pdf",
+ )
+ db_session.add(file_record)
+ db_session.commit()
+ db_session.refresh(file_record)
+
+ response = client.post(
+ f"/api/files/{file_record.id}/retry-subtask?subtask_name=process_with_azure_document_intelligence"
+ )
+ assert response.status_code == 400
+ assert "not found on disk" in response.json()["detail"].lower()
+
@pytest.mark.integration
class TestFilePreview:
diff --git a/tests/test_process_document.py b/tests/test_process_document.py
index 84112867..6ce6971c 100644
--- a/tests/test_process_document.py
+++ b/tests/test_process_document.py
@@ -6,12 +6,13 @@ and doesn't cause DetachedInstanceError when accessing database objects.
"""
import os
+from unittest.mock import MagicMock, patch
+
import pytest
-from unittest.mock import patch, MagicMock
from sqlalchemy.orm import Session
-from app.tasks.process_document import process_document
from app.models import FileRecord
+from app.tasks.process_document import process_document
@pytest.mark.unit
@@ -85,11 +86,12 @@ startxref
test_pdf.write_bytes(pdf_content)
# Mock environment and dependencies
- 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:
+ 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,
+ ):
# Setup mocks
mock_settings.workdir = str(tmp_path)
@@ -147,8 +149,9 @@ def test_process_document_duplicate_file(db_session, tmp_path):
existing_id = existing_record.id
# Mock environment and dependencies
- with patch("app.tasks.process_document.SessionLocal") as mock_session_local, patch(
- "app.tasks.process_document.log_task_progress"
+ with (
+ patch("app.tasks.process_document.SessionLocal") as mock_session_local,
+ patch("app.tasks.process_document.log_task_progress"),
):
# Setup mocks
@@ -213,11 +216,12 @@ startxref
test_pdf.write_bytes(pdf_content)
# Mock environment and dependencies
- 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.process_with_azure_document_intelligence"
- ) as mock_azure:
+ 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.process_with_azure_document_intelligence") as mock_azure,
+ ):
# Setup mocks
mock_settings.workdir = str(tmp_path)
@@ -242,3 +246,142 @@ startxref
# The second argument should be the file_id
assert call_args[0][1] == file_record.id
+
+
+@pytest.mark.unit
+@pytest.mark.requires_db
+def test_process_document_reprocess_skips_duplicate_check(db_session, tmp_path):
+ """
+ Test that reprocessing an existing file (with file_id) skips the duplicate check
+ and continues processing normally.
+ """
+ # 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)
+
+ # Pre-create a FileRecord with the same hash (simulating an existing record)
+ from app.utils import hash_file
+
+ filehash = hash_file(str(test_pdf))
+
+ existing_record = FileRecord(
+ filehash=filehash,
+ original_filename="test.pdf",
+ local_filename=str(test_pdf),
+ file_size=len(pdf_content),
+ mime_type="application/pdf",
+ )
+ db_session.add(existing_record)
+ db_session.commit()
+ existing_id = existing_record.id
+
+ # Mock environment and dependencies
+ 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,
+ ):
+
+ # Setup mocks
+ mock_settings.workdir = str(tmp_path)
+ mock_session_local.return_value.__enter__.return_value = db_session
+ mock_session_local.return_value.__exit__.return_value = None
+ mock_extract.delay = MagicMock()
+
+ # Call with file_id to trigger reprocessing (should skip duplicate check)
+ result = process_document.run(str(test_pdf), file_id=existing_id)
+
+ # Verify that processing continued (not blocked by duplicate check)
+ assert result["status"] == "Text extracted locally"
+ assert result["file_id"] == existing_id
+
+ # Verify that extract_metadata_with_gpt was called
+ mock_extract.delay.assert_called_once()
+
+ # Verify that only one FileRecord still exists (no new record created)
+ assert db_session.query(FileRecord).count() == 1
+
+
+@pytest.mark.unit
+@pytest.mark.requires_db
+def test_process_document_reprocess_nonexistent_file_id(db_session, tmp_path):
+ """
+ Test that reprocessing with a non-existent file_id returns an error.
+ """
+ # Create a test PDF file
+ test_pdf = tmp_path / "test.pdf"
+ test_pdf.write_bytes(b"test content")
+
+ with (
+ patch("app.tasks.process_document.SessionLocal") as mock_session_local,
+ patch("app.tasks.process_document.log_task_progress"),
+ ):
+ mock_session_local.return_value.__enter__.return_value = db_session
+ mock_session_local.return_value.__exit__.return_value = None
+
+ # Call with a file_id that doesn't exist
+ result = process_document.run(str(test_pdf), file_id=99999)
+
+ # Verify error is returned
+ assert "error" in result
+ assert result["file_id"] == 99999