fix: handle file in processed directory when retrying embed_metadata_into_pdf

- Update _retry_pipeline_step to check for file in tmp, processed, and fallback locations
- Pass full path to extract_metadata_with_gpt instead of just basename
- Update extract_metadata_with_gpt to handle both basename and full path parameters
- Add test case for retrying when file is in processed directory

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>

app/api/files.py

@@ -525,13 +525,35 @@ def _retry_pipeline_step(file_record: FileRecord, step_name: str, db: Session) -
         # Retrying embed requires re-running metadata extraction first, because
         # embed_metadata_into_pdf needs the actual metadata dict (not empty).
         # Re-trigger extract_metadata_with_gpt which will chain into embed_metadata_into_pdf.
-        if not file_record.local_filename or not os.path.exists(file_record.local_filename):
+        
+        # Check for file in multiple locations:
+        # 1. Original location in tmp (file_record.local_filename)
+        # 2. Processed location (file_record.processed_file_path)
+        # 3. Fallback to workdir/tmp/<basename>
+        file_path = None
+        if file_record.local_filename and os.path.exists(file_record.local_filename):
+            file_path = file_record.local_filename
+        elif file_record.processed_file_path and os.path.exists(file_record.processed_file_path):
+            # File has been processed and moved to processed directory
+            file_path = file_record.processed_file_path
+        else:
+            # Try fallback path in workdir/tmp
+            if file_record.local_filename:
+                workdir = settings.workdir
+                tmp_dir = os.path.join(workdir, "tmp")
+                fallback_path = os.path.join(tmp_dir, os.path.basename(file_record.local_filename))
+                if os.path.exists(fallback_path):
+                    file_path = fallback_path
+        
+        if not file_path:
             raise HTTPException(
-                status_code=400, detail="Local file not found on disk. Cannot retry metadata embedding."
+                status_code=400, 
+                detail="File not found in tmp or processed directory. Cannot retry metadata embedding."
             )
-        extracted_text = _extract_text_from_pdf(file_record.local_filename)
-        filename = os.path.basename(file_record.local_filename)
-        task = extract_metadata_task.delay(filename, extracted_text, file_id)
+        
+        extracted_text = _extract_text_from_pdf(file_path)
+        # Pass the full path to the task so it can locate the file
+        task = extract_metadata_task.delay(file_path, extracted_text, file_id)
     else:
         raise HTTPException(status_code=400, detail=f"Unsupported pipeline step: {step_name}")
 

app/tasks/extract_metadata_with_gpt.py

@@ -47,17 +47,28 @@ def extract_json_from_text(text):
 
 @celery.task(base=BaseTaskWithRetry, bind=True)
 def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: int = None):
-    """Uses OpenAI to classify document metadata."""
+    """
+    Uses OpenAI to classify document metadata.
+    
+    Args:
+        filename: Can be either a basename (e.g., "file.pdf") or a full path (e.g., "/workdir/processed/file.pdf")
+        cleaned_text: The extracted text from the document
+        file_id: Optional file ID for tracking
+    """
     task_id = self.request.id
     logger.info(f"[{task_id}] Starting metadata extraction for: {filename}")
     log_task_progress(
-        task_id, "extract_metadata_with_gpt", "in_progress", f"Extracting metadata for {filename}", file_id=file_id
+        task_id, "extract_metadata_with_gpt", "in_progress", f"Extracting metadata for {os.path.basename(filename)}", file_id=file_id
     )
 
     # Get file_id from database if not provided
     if file_id is None:
         tmp_dir = os.path.join(settings.workdir, "tmp")
-        file_path = os.path.join(tmp_dir, filename)
+        # Handle both basename and full path
+        if os.path.isabs(filename):
+            file_path = filename
+        else:
+            file_path = os.path.join(tmp_dir, filename)
         if os.path.exists(file_path):
             with SessionLocal() as db:
                 file_record = db.query(FileRecord).filter_by(local_filename=file_path).first()
@@ -162,14 +173,14 @@ def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: i
         )
 
         # Trigger the next step: embedding metadata into the PDF
-        # Pass the original filename (UUID-based) so embed_metadata_into_pdf can find the file on disk
+        # Pass the filename (can be basename or full path) so embed_metadata_into_pdf can find the file on disk
         logger.info(f"[{task_id}] Queueing metadata embedding task")
         log_task_progress(
             task_id, "extract_metadata_with_gpt", "success", "Metadata extracted, queuing embed task", file_id=file_id
         )
         embed_metadata_into_pdf.delay(filename, cleaned_text, metadata, file_id)
 
-        return {"s3_file": filename, "metadata": metadata}
+        return {"s3_file": os.path.basename(filename), "metadata": metadata}
 
     except Exception as e:
         logger.exception(f"[{task_id}] OpenAI classification failed for {filename}: {e}")