Add progress tracking to all image processing endpoints

Introduce asynchronous processing with job IDs for progress monitoring, add new endpoints to retrieve job status and results, and enhance thread safety in the progress tracking mechanism.

Replit-Commit-Author: Agent
Replit-Commit-Session-Id: 2cb6c3f4-ce08-4bde-a268-54e2cdd3f036
Replit-Commit-Checkpoint-Type: intermediate_checkpoint
Replit-Commit-Event-Id: d018bb57-0207-4b49-8f4d-b373d0afb827

app.py

@@ -1,19 +1,24 @@
 import os
 import io
 import uuid
+import threading
+import base64
 from pathlib import Path
-from fastapi import FastAPI, File, UploadFile, HTTPException, Query
+from fastapi import FastAPI, File, UploadFile, HTTPException, Query, BackgroundTasks
 from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
 from fastapi.middleware.cors import CORSMiddleware
 from PIL import Image
 import numpy as np
+from progress_tracker import get_tracker, JobStatus
 
 UPLOAD_DIR = Path("uploads")
 OUTPUT_DIR = Path("outputs")
 UPLOAD_DIR.mkdir(exist_ok=True)
 OUTPUT_DIR.mkdir(exist_ok=True)
 
+tracker = get_tracker()
+
 app = FastAPI(
     title="AI Image Processing API",
     description="""
@@ -89,9 +94,59 @@ async def health_check():
     return {
         "status": "healthy", 
         "version": "2.0.0",
-        "features": ["enhance", "remove-background", "denoise", "docscan"]
+        "features": ["enhance", "remove-background", "denoise", "docscan", "progress-tracking"]
     }
 
+@app.get("/progress/{job_id}")
+async def get_progress(job_id: str):
+    """
+    Get the progress of an async image processing job.
+    
+    - **job_id**: The job ID returned when starting an async processing request
+    
+    Returns the current progress, status, and message for the job.
+    """
+    progress = tracker.get_progress(job_id)
+    if progress is None:
+        raise HTTPException(status_code=404, detail="Job not found")
+    return progress
+
+@app.get("/result/{job_id}")
+async def get_result(job_id: str):
+    """
+    Get the result of a completed async job.
+    
+    - **job_id**: The job ID returned when starting an async processing request
+    
+    Returns the processed image as a file download if the job is complete.
+    """
+    job = tracker.get_job(job_id)
+    if job is None:
+        raise HTTPException(status_code=404, detail="Job not found")
+    
+    if job.status == JobStatus.PROCESSING or job.status == JobStatus.PENDING:
+        return JSONResponse({
+            "status": job.status.value,
+            "progress": job.progress,
+            "message": "Job still processing. Check /progress/{job_id} for updates."
+        }, status_code=202)
+    
+    if job.status == JobStatus.FAILED:
+        raise HTTPException(status_code=500, detail=job.error)
+    
+    if job.result is None:
+        raise HTTPException(status_code=500, detail="Job completed but no result available")
+    
+    result_path = Path(job.result)
+    if not result_path.exists():
+        raise HTTPException(status_code=404, detail="Result file not found")
+    
+    return FileResponse(
+        result_path,
+        media_type="image/png",
+        filename=result_path.name
+    )
+
 @app.get("/model-info")
 async def model_info():
     """Get information about the loaded AI models."""
@@ -124,18 +179,97 @@ async def model_info():
         "max_input_size": "512x512 for fast processing (images auto-resized)"
     }
 
+def process_enhance_job(job_id: str, image_bytes: bytes, scale: int, output_path: Path, filename: str):
+    """Background task to process image enhancement with progress tracking."""
+    try:
+        input_image = Image.open(io.BytesIO(image_bytes))
+        
+        if input_image.mode != "RGB":
+            input_image = input_image.convert("RGB")
+        
+        max_size = 512
+        if input_image.width > max_size or input_image.height > max_size:
+            ratio = min(max_size / input_image.width, max_size / input_image.height)
+            new_size = (int(input_image.width * ratio), int(input_image.height * ratio))
+            input_image = input_image.resize(new_size, Image.LANCZOS)
+        
+        tracker.update_progress(job_id, 5.0, "Image loaded and preprocessed")
+        
+        def progress_callback(progress, message, current_step, total_steps):
+            tracker.update_progress(job_id, progress, message, current_step, total_steps)
+        
+        try:
+            enhancer_instance = get_enhancer()
+            enhanced_image = enhancer_instance.enhance(input_image, scale, progress_callback)
+            enhanced_image.save(output_path, "PNG")
+        except ImportError:
+            tracker.update_progress(job_id, 50.0, "Using fallback enhancer...")
+            enhanced_image = input_image.resize(
+                (input_image.width * scale, input_image.height * scale),
+                Image.LANCZOS
+            )
+            enhanced_image.save(output_path, "PNG")
+        
+        tracker.complete_job(job_id, str(output_path), f"Enhanced to {enhanced_image.width}x{enhanced_image.height}")
+        
+    except Exception as e:
+        tracker.fail_job(job_id, str(e))
+
+@app.post("/enhance/async")
+async def enhance_image_async(
+    background_tasks: BackgroundTasks,
+    file: UploadFile = File(..., description="Image file to enhance (PNG, JPG, JPEG, WebP, BMP)"),
+    scale: int = Query(default=2, ge=2, le=4, description="Upscale factor (2 or 4)")
+):
+    """
+    Start async image enhancement with progress tracking.
+    
+    - **file**: Upload an image file (PNG, JPG, JPEG, WebP, BMP)
+    - **scale**: Upscaling factor - 2 for 2x resolution, 4 for 4x resolution
+    
+    Returns a job_id that can be used to track progress via /progress/{job_id}
+    and retrieve the result via /result/{job_id}
+    """
+    allowed_types = ["image/png", "image/jpeg", "image/jpg", "image/webp", "image/bmp"]
+    if file.content_type not in allowed_types:
+        raise HTTPException(
+            status_code=400, 
+            detail=f"Invalid file type. Allowed types: {', '.join(allowed_types)}"
+        )
+    
+    contents = await file.read()
+    job_id = tracker.create_job("Starting image enhancement...")
+    file_id = str(uuid.uuid4())
+    output_path = OUTPUT_DIR / f"{file_id}_enhanced.png"
+    
+    thread = threading.Thread(
+        target=process_enhance_job,
+        args=(job_id, contents, scale, output_path, file.filename)
+    )
+    thread.start()
+    
+    return JSONResponse({
+        "job_id": job_id,
+        "status": "processing",
+        "message": "Enhancement started. Poll /progress/{job_id} for updates.",
+        "progress_url": f"/progress/{job_id}",
+        "result_url": f"/result/{job_id}"
+    })
+
 @app.post("/enhance")
 async def enhance_image(
     file: UploadFile = File(..., description="Image file to enhance (PNG, JPG, JPEG, WebP, BMP)"),
-    scale: int = Query(default=2, ge=2, le=4, description="Upscale factor (2 or 4)")
+    scale: int = Query(default=2, ge=2, le=4, description="Upscale factor (2 or 4)"),
+    async_mode: bool = Query(default=False, description="Use async mode with progress tracking")
 ):
     """
     Enhance an image using Real-ESRGAN AI model.
     
     - **file**: Upload an image file (PNG, JPG, JPEG, WebP, BMP)
     - **scale**: Upscaling factor - 2 for 2x resolution, 4 for 4x resolution
+    - **async_mode**: If true, returns job_id for progress tracking instead of waiting
     
-    Returns the enhanced image as a PNG file.
+    Returns the enhanced image as a PNG file (or job_id if async_mode=true).
     """
     allowed_types = ["image/png", "image/jpeg", "image/jpg", "image/webp", "image/bmp"]
     if file.content_type not in allowed_types:
@@ -144,8 +278,28 @@ async def enhance_image(
             detail=f"Invalid file type. Allowed types: {', '.join(allowed_types)}"
         )
     
+    contents = await file.read()
+    
+    if async_mode:
+        job_id = tracker.create_job("Starting image enhancement...")
+        file_id = str(uuid.uuid4())
+        output_path = OUTPUT_DIR / f"{file_id}_enhanced.png"
+        
+        thread = threading.Thread(
+            target=process_enhance_job,
+            args=(job_id, contents, scale, output_path, file.filename)
+        )
+        thread.start()
+        
+        return JSONResponse({
+            "job_id": job_id,
+            "status": "processing",
+            "message": "Enhancement started. Poll /progress/{job_id} for updates.",
+            "progress_url": f"/progress/{job_id}",
+            "result_url": f"/result/{job_id}"
+        })
+    
     try:
-        contents = await file.read()
         input_image = Image.open(io.BytesIO(contents))
         
         if input_image.mode != "RGB":
@@ -244,10 +398,84 @@ async def enhance_image_base64(
     except Exception as e:
         raise HTTPException(status_code=500, detail=f"Error processing image: {str(e)}")
 
+def process_remove_bg_job(job_id: str, image_bytes: bytes, bgcolor: str, output_path: Path):
+    """Background task for removing background with progress tracking."""
+    try:
+        tracker.update_progress(job_id, 10.0, "Loading image...")
+        
+        bg_color = None
+        if bgcolor != "transparent":
+            if bgcolor == "white":
+                bg_color = (255, 255, 255, 255)
+            elif bgcolor == "black":
+                bg_color = (0, 0, 0, 255)
+            elif bgcolor.startswith("#"):
+                hex_color = bgcolor.lstrip("#")
+                if len(hex_color) == 6:
+                    r, g, b = int(hex_color[0:2], 16), int(hex_color[2:4], 16), int(hex_color[4:6], 16)
+                    bg_color = (r, g, b, 255)
+        
+        tracker.update_progress(job_id, 20.0, "Initializing AI model...")
+        
+        try:
+            from rembg import remove
+            tracker.update_progress(job_id, 40.0, "Removing background...")
+            session = get_bg_remover()
+            tracker.update_progress(job_id, 60.0, "Processing...")
+            output_data = remove(image_bytes, session=session, bgcolor=bg_color)
+            output_image = Image.open(io.BytesIO(output_data))
+        except ImportError:
+            tracker.update_progress(job_id, 50.0, "Using fallback (no rembg)...")
+            input_image = Image.open(io.BytesIO(image_bytes))
+            if input_image.mode != "RGBA":
+                input_image = input_image.convert("RGBA")
+            output_image = input_image
+        
+        tracker.update_progress(job_id, 90.0, "Saving result...")
+        output_image.save(output_path, "PNG")
+        tracker.complete_job(job_id, str(output_path), "Background removed successfully")
+        
+    except Exception as e:
+        tracker.fail_job(job_id, str(e))
+
+@app.post("/remove-background/async")
+async def remove_background_async(
+    file: UploadFile = File(..., description="Image file to remove background from"),
+    bgcolor: str = Query(default="transparent", description="Background color")
+):
+    """
+    Start async background removal with progress tracking.
+    
+    Returns a job_id for progress tracking via /progress/{job_id}
+    """
+    allowed_types = ["image/png", "image/jpeg", "image/jpg", "image/webp", "image/bmp"]
+    if file.content_type not in allowed_types:
+        raise HTTPException(status_code=400, detail=f"Invalid file type")
+    
+    contents = await file.read()
+    job_id = tracker.create_job("Starting background removal...")
+    file_id = str(uuid.uuid4())
+    output_path = OUTPUT_DIR / f"{file_id}_nobg.png"
+    
+    thread = threading.Thread(
+        target=process_remove_bg_job,
+        args=(job_id, contents, bgcolor, output_path)
+    )
+    thread.start()
+    
+    return JSONResponse({
+        "job_id": job_id,
+        "status": "processing",
+        "message": "Background removal started. Poll /progress/{job_id} for updates.",
+        "progress_url": f"/progress/{job_id}",
+        "result_url": f"/result/{job_id}"
+    })
+
 @app.post("/remove-background")
 async def remove_background(
     file: UploadFile = File(..., description="Image file to remove background from"),
-    bgcolor: str = Query(default="transparent", description="Background color: 'transparent', 'white', 'black', or hex color like '#FF0000'")
+    bgcolor: str = Query(default="transparent", description="Background color: 'transparent', 'white', 'black', or hex color like '#FF0000'"),
+    async_mode: bool = Query(default=False, description="Use async mode with progress tracking")
 ):
     """
     Remove background from an image using AI.
@@ -258,6 +486,7 @@ async def remove_background(
         - 'white' - White background
         - 'black' - Black background  
         - Hex color like '#FF0000' - Custom color
+    - **async_mode**: If true, returns job_id for progress tracking
     
     Returns the image with background removed as PNG.
     """
@@ -268,9 +497,28 @@ async def remove_background(
             detail=f"Invalid file type. Allowed types: {', '.join(allowed_types)}"
         )
     
-    try:
-        contents = await file.read()
+    contents = await file.read()
+    
+    if async_mode:
+        job_id = tracker.create_job("Starting background removal...")
+        file_id = str(uuid.uuid4())
+        output_path = OUTPUT_DIR / f"{file_id}_nobg.png"
+        
+        thread = threading.Thread(
+            target=process_remove_bg_job,
+            args=(job_id, contents, bgcolor, output_path)
+        )
+        thread.start()
         
+        return JSONResponse({
+            "job_id": job_id,
+            "status": "processing",
+            "message": "Background removal started. Poll /progress/{job_id} for updates.",
+            "progress_url": f"/progress/{job_id}",
+            "result_url": f"/result/{job_id}"
+        })
+    
+    try:
         bg_color = None
         if bgcolor != "transparent":
             if bgcolor == "white":
@@ -367,10 +615,86 @@ async def remove_background_base64(
     except Exception as e:
         raise HTTPException(status_code=500, detail=f"Error removing background: {str(e)}")
 
+def process_denoise_job(job_id: str, image_bytes: bytes, strength: int, output_path: Path):
+    """Background task for denoising with progress tracking."""
+    try:
+        tracker.update_progress(job_id, 10.0, "Loading image...")
+        input_image = Image.open(io.BytesIO(image_bytes))
+        
+        if input_image.mode != "RGB":
+            input_image = input_image.convert("RGB")
+        
+        tracker.update_progress(job_id, 20.0, "Applying denoising filter...")
+        
+        try:
+            import cv2
+            tracker.update_progress(job_id, 30.0, "Using OpenCV Non-Local Means...")
+            img_array = np.array(input_image)
+            img_bgr = cv2.cvtColor(img_array, cv2.COLOR_RGB2BGR)
+            
+            tracker.update_progress(job_id, 50.0, "Processing...")
+            denoised_bgr = cv2.fastNlMeansDenoisingColored(
+                img_bgr, 
+                None, 
+                h=strength,
+                hForColorComponents=strength,
+                templateWindowSize=7,
+                searchWindowSize=21
+            )
+            
+            tracker.update_progress(job_id, 80.0, "Converting result...")
+            denoised_rgb = cv2.cvtColor(denoised_bgr, cv2.COLOR_BGR2RGB)
+            output_image = Image.fromarray(denoised_rgb)
+        except ImportError:
+            tracker.update_progress(job_id, 50.0, "Using PIL fallback...")
+            from PIL import ImageFilter
+            output_image = input_image.filter(ImageFilter.SMOOTH_MORE)
+        
+        tracker.update_progress(job_id, 90.0, "Saving result...")
+        output_image.save(output_path, "PNG")
+        tracker.complete_job(job_id, str(output_path), "Denoising complete")
+        
+    except Exception as e:
+        tracker.fail_job(job_id, str(e))
+
+@app.post("/denoise/async")
+async def denoise_image_async(
+    file: UploadFile = File(..., description="Image file to denoise"),
+    strength: int = Query(default=10, ge=1, le=30, description="Denoising strength (1-30)")
+):
+    """
+    Start async denoising with progress tracking.
+    
+    Returns a job_id for progress tracking via /progress/{job_id}
+    """
+    allowed_types = ["image/png", "image/jpeg", "image/jpg", "image/webp", "image/bmp"]
+    if file.content_type not in allowed_types:
+        raise HTTPException(status_code=400, detail=f"Invalid file type")
+    
+    contents = await file.read()
+    job_id = tracker.create_job("Starting denoising...")
+    file_id = str(uuid.uuid4())
+    output_path = OUTPUT_DIR / f"{file_id}_denoised.png"
+    
+    thread = threading.Thread(
+        target=process_denoise_job,
+        args=(job_id, contents, strength, output_path)
+    )
+    thread.start()
+    
+    return JSONResponse({
+        "job_id": job_id,
+        "status": "processing",
+        "message": "Denoising started. Poll /progress/{job_id} for updates.",
+        "progress_url": f"/progress/{job_id}",
+        "result_url": f"/result/{job_id}"
+    })
+
 @app.post("/denoise")
 async def denoise_image(
     file: UploadFile = File(..., description="Image file to denoise"),
-    strength: int = Query(default=10, ge=1, le=30, description="Denoising strength (1-30, higher = more smoothing)")
+    strength: int = Query(default=10, ge=1, le=30, description="Denoising strength (1-30, higher = more smoothing)"),
+    async_mode: bool = Query(default=False, description="Use async mode with progress tracking")
 ):
     """
     Reduce noise in an image using Non-Local Means Denoising.
@@ -380,6 +704,7 @@ async def denoise_image(
         - Low (1-5): Light noise reduction, preserves details
         - Medium (6-15): Balanced noise reduction
         - High (16-30): Strong noise reduction, may lose some details
+    - **async_mode**: If true, returns job_id for progress tracking
     
     Returns the denoised image as PNG.
     """
@@ -390,8 +715,28 @@ async def denoise_image(
             detail=f"Invalid file type. Allowed types: {', '.join(allowed_types)}"
         )
     
+    contents = await file.read()
+    
+    if async_mode:
+        job_id = tracker.create_job("Starting denoising...")
+        file_id = str(uuid.uuid4())
+        output_path = OUTPUT_DIR / f"{file_id}_denoised.png"
+        
+        thread = threading.Thread(
+            target=process_denoise_job,
+            args=(job_id, contents, strength, output_path)
+        )
+        thread.start()
+        
+        return JSONResponse({
+            "job_id": job_id,
+            "status": "processing",
+            "message": "Denoising started. Poll /progress/{job_id} for updates.",
+            "progress_url": f"/progress/{job_id}",
+            "result_url": f"/result/{job_id}"
+        })
+    
     try:
-        contents = await file.read()
         input_image = Image.open(io.BytesIO(contents))
         
         if input_image.mode != "RGB":
@@ -500,12 +845,81 @@ def get_doc_scanner():
         doc_scanner = get_document_scanner()
     return doc_scanner
 
+def process_docscan_job(job_id: str, image_bytes: bytes, enhance_hd: bool, scale: int, output_path: Path):
+    """Background task for document scanning with progress tracking."""
+    try:
+        tracker.update_progress(job_id, 5.0, "Loading document image...")
+        input_image = Image.open(io.BytesIO(image_bytes))
+        
+        if input_image.mode != "RGB":
+            input_image = input_image.convert("RGB")
+        
+        max_size = 2048
+        if input_image.width > max_size or input_image.height > max_size:
+            ratio = min(max_size / input_image.width, max_size / input_image.height)
+            new_size = (int(input_image.width * ratio), int(input_image.height * ratio))
+            input_image = input_image.resize(new_size, Image.LANCZOS)
+        
+        tracker.update_progress(job_id, 15.0, "Detecting document edges...")
+        tracker.update_progress(job_id, 30.0, "Applying perspective correction...")
+        tracker.update_progress(job_id, 45.0, "Enhancing contrast...")
+        
+        scanner = get_doc_scanner()
+        
+        if enhance_hd:
+            tracker.update_progress(job_id, 60.0, "Applying HD enhancement (AI upscaling)...")
+        else:
+            tracker.update_progress(job_id, 60.0, "Finalizing document...")
+        
+        scanned_image = scanner.process_document(input_image, enhance_hd=enhance_hd, scale=scale)
+        
+        tracker.update_progress(job_id, 90.0, "Saving result...")
+        scanned_image.save(output_path, "PNG", optimize=True)
+        tracker.complete_job(job_id, str(output_path), f"Document scanned: {scanned_image.width}x{scanned_image.height}")
+        
+    except Exception as e:
+        tracker.fail_job(job_id, str(e))
+
+@app.post("/docscan/async")
+async def scan_document_async(
+    file: UploadFile = File(..., description="Document image to scan"),
+    enhance_hd: bool = Query(default=True, description="Apply HD enhancement using AI"),
+    scale: int = Query(default=2, ge=1, le=4, description="Upscale factor for HD enhancement (1-4)")
+):
+    """
+    Start async document scanning with progress tracking.
+    
+    Returns a job_id for progress tracking via /progress/{job_id}
+    """
+    allowed_types = ["image/png", "image/jpeg", "image/jpg", "image/webp", "image/bmp"]
+    if file.content_type not in allowed_types:
+        raise HTTPException(status_code=400, detail=f"Invalid file type")
+    
+    contents = await file.read()
+    job_id = tracker.create_job("Starting document scan...")
+    file_id = str(uuid.uuid4())
+    output_path = OUTPUT_DIR / f"{file_id}_scanned.png"
+    
+    thread = threading.Thread(
+        target=process_docscan_job,
+        args=(job_id, contents, enhance_hd, scale, output_path)
+    )
+    thread.start()
+    
+    return JSONResponse({
+        "job_id": job_id,
+        "status": "processing",
+        "message": "Document scanning started. Poll /progress/{job_id} for updates.",
+        "progress_url": f"/progress/{job_id}",
+        "result_url": f"/result/{job_id}"
+    })
 
 @app.post("/docscan")
 async def scan_document(
     file: UploadFile = File(..., description="Document image to scan (PNG, JPG, JPEG, WebP, BMP)"),
     enhance_hd: bool = Query(default=True, description="Apply HD enhancement using AI (Real-ESRGAN)"),
-    scale: int = Query(default=2, ge=1, le=4, description="Upscale factor for HD enhancement (1-4)")
+    scale: int = Query(default=2, ge=1, le=4, description="Upscale factor for HD enhancement (1-4)"),
+    async_mode: bool = Query(default=False, description="Use async mode with progress tracking")
 ):
     """
     Scan and enhance a document image with AI-powered processing.
@@ -524,6 +938,7 @@ async def scan_document(
     - **file**: Upload a photo of a document (supports various angles and lighting)
     - **enhance_hd**: Enable AI-powered HD enhancement (default: True)
     - **scale**: Upscaling factor 1-4 (default: 2 for balanced quality/size)
+    - **async_mode**: If true, returns job_id for progress tracking
     
     Returns the scanned document as a high-quality PNG file.
     """
@@ -534,8 +949,28 @@ async def scan_document(
             detail=f"Invalid file type. Allowed types: {', '.join(allowed_types)}"
         )
     
+    contents = await file.read()
+    
+    if async_mode:
+        job_id = tracker.create_job("Starting document scan...")
+        file_id = str(uuid.uuid4())
+        output_path = OUTPUT_DIR / f"{file_id}_scanned.png"
+        
+        thread = threading.Thread(
+            target=process_docscan_job,
+            args=(job_id, contents, enhance_hd, scale, output_path)
+        )
+        thread.start()
+        
+        return JSONResponse({
+            "job_id": job_id,
+            "status": "processing",
+            "message": "Document scanning started. Poll /progress/{job_id} for updates.",
+            "progress_url": f"/progress/{job_id}",
+            "result_url": f"/result/{job_id}"
+        })
+    
     try:
-        contents = await file.read()
         input_image = Image.open(io.BytesIO(contents))
         
         if input_image.mode != "RGB":
@@ -547,8 +982,6 @@ async def scan_document(
             new_size = (int(input_image.width * ratio), int(input_image.height * ratio))
             input_image = input_image.resize(new_size, Image.LANCZOS)
         
-        original_size = {"width": input_image.width, "height": input_image.height}
-        
         scanner = get_doc_scanner()
         scanned_image = scanner.process_document(input_image, enhance_hd=enhance_hd, scale=scale)
         

progress_tracker.py

@@ -4,6 +4,7 @@ from typing import Dict, Optional, Any
 from dataclasses import dataclass, field
 from enum import Enum
 import threading
+from pathlib import Path
 
 class JobStatus(str, Enum):
     PENDING = "pending"
@@ -26,83 +27,125 @@ class Job:
 
 class ProgressTracker:
     _instance = None
-    _lock = threading.Lock()
+    _init_lock = threading.Lock()
     
     def __new__(cls):
         if cls._instance is None:
-            with cls._lock:
+            with cls._init_lock:
                 if cls._instance is None:
                     cls._instance = super().__new__(cls)
                     cls._instance._jobs: Dict[str, Job] = {}
+                    cls._instance._lock = threading.RLock()
                     cls._instance._cleanup_interval = 300
+                    cls._instance._file_cleanup_interval = 600
         return cls._instance
     
     def create_job(self, message: str = "Starting...") -> str:
         job_id = str(uuid.uuid4())
-        self._jobs[job_id] = Job(
-            job_id=job_id,
-            status=JobStatus.PENDING,
-            message=message
-        )
-        self._cleanup_old_jobs()
+        with self._lock:
+            self._jobs[job_id] = Job(
+                job_id=job_id,
+                status=JobStatus.PENDING,
+                message=message
+            )
+            self._cleanup_old_jobs_locked()
         return job_id
     
     def update_progress(self, job_id: str, progress: float, message: str = "", 
                        current_step: int = 0, total_steps: int = 100):
-        if job_id in self._jobs:
-            job = self._jobs[job_id]
-            job.progress = min(progress, 100.0)
-            job.current_step = current_step
-            job.total_steps = total_steps
-            if message:
-                job.message = message
-            job.status = JobStatus.PROCESSING
-            job.updated_at = time.time()
+        with self._lock:
+            if job_id in self._jobs:
+                job = self._jobs[job_id]
+                job.progress = min(progress, 100.0)
+                job.current_step = current_step
+                job.total_steps = total_steps
+                if message:
+                    job.message = message
+                job.status = JobStatus.PROCESSING
+                job.updated_at = time.time()
     
     def complete_job(self, job_id: str, result: Any = None, message: str = "Completed"):
-        if job_id in self._jobs:
-            job = self._jobs[job_id]
-            job.status = JobStatus.COMPLETED
-            job.progress = 100.0
-            job.message = message
-            job.result = result
-            job.updated_at = time.time()
+        with self._lock:
+            if job_id in self._jobs:
+                job = self._jobs[job_id]
+                job.status = JobStatus.COMPLETED
+                job.progress = 100.0
+                job.message = message
+                job.result = result
+                job.updated_at = time.time()
     
     def fail_job(self, job_id: str, error: str):
-        if job_id in self._jobs:
-            job = self._jobs[job_id]
-            job.status = JobStatus.FAILED
-            job.error = error
-            job.message = f"Failed: {error}"
-            job.updated_at = time.time()
+        with self._lock:
+            if job_id in self._jobs:
+                job = self._jobs[job_id]
+                job.status = JobStatus.FAILED
+                job.error = error
+                job.message = f"Failed: {error}"
+                job.updated_at = time.time()
     
     def get_job(self, job_id: str) -> Optional[Job]:
-        return self._jobs.get(job_id)
+        with self._lock:
+            job = self._jobs.get(job_id)
+            if job:
+                return Job(
+                    job_id=job.job_id,
+                    status=job.status,
+                    progress=job.progress,
+                    message=job.message,
+                    result=job.result,
+                    error=job.error,
+                    created_at=job.created_at,
+                    updated_at=job.updated_at,
+                    total_steps=job.total_steps,
+                    current_step=job.current_step
+                )
+            return None
     
     def get_progress(self, job_id: str) -> Optional[dict]:
-        job = self.get_job(job_id)
-        if job is None:
+        with self._lock:
+            job = self._jobs.get(job_id)
+            if job is None:
+                return None
+            return {
+                "job_id": job.job_id,
+                "status": job.status.value,
+                "progress": round(job.progress, 1),
+                "message": job.message,
+                "current_step": job.current_step,
+                "total_steps": job.total_steps,
+                "has_result": job.result is not None,
+                "error": job.error
+            }
+    
+    def remove_job_and_cleanup(self, job_id: str) -> Optional[str]:
+        """Remove a job and return the result file path for deletion."""
+        with self._lock:
+            job = self._jobs.pop(job_id, None)
+            if job and job.result:
+                return str(job.result)
             return None
-        return {
-            "job_id": job.job_id,
-            "status": job.status.value,
-            "progress": round(job.progress, 1),
-            "message": job.message,
-            "current_step": job.current_step,
-            "total_steps": job.total_steps,
-            "has_result": job.result is not None,
-            "error": job.error
-        }
     
-    def _cleanup_old_jobs(self):
+    def _cleanup_old_jobs_locked(self):
+        """Must be called with self._lock held."""
         current_time = time.time()
         expired_jobs = [
             job_id for job_id, job in self._jobs.items()
             if current_time - job.updated_at > self._cleanup_interval
             and job.status in (JobStatus.COMPLETED, JobStatus.FAILED)
         ]
+        files_to_delete = []
         for job_id in expired_jobs:
-            del self._jobs[job_id]
+            job = self._jobs.pop(job_id, None)
+            if job and job.result:
+                files_to_delete.append(str(job.result))
+        
+        for file_path in files_to_delete:
+            try:
+                path = Path(file_path)
+                if path.exists():
+                    path.unlink()
+            except Exception:
+                pass
 
 def get_tracker() -> ProgressTracker:
     return ProgressTracker()