new worker

core/models.py

@@ -103,6 +103,7 @@ class AuditLog(Base):
 class GeminiJob(Base):
     """
     Generic job queue for Gemini operations (video, image, text).
+    Uses priority-tier system for worker assignment.
     """
     __tablename__ = "gemini_jobs"
 
@@ -111,7 +112,13 @@ class GeminiJob(Base):
     user_id = Column(String(50), index=True, nullable=False)  # User who requested
     job_type = Column(String(20), index=True, nullable=False)  # video, image, text, analyze
     third_party_id = Column(String(255), nullable=True)  # Gemini operation name (for video)
-    status = Column(String(20), default="queued", index=True)  # queued, processing, completed, failed
+    status = Column(String(20), default="queued", index=True)  # queued, processing, completed, failed, cancelled
+    
+    # Priority-tier worker system
+    priority = Column(String(10), default="fast", index=True)  # fast (5s), medium (30s), slow (60s)
+    next_process_at = Column(DateTime(timezone=True), nullable=True, index=True)  # When worker should pick up again
+    retry_count = Column(Integer, default=0)  # Number of status check retries
+    
     input_data = Column(JSON, nullable=True)  # Request details (prompt, settings, etc.)
     output_data = Column(JSON, nullable=True)  # Result (filename, text, etc.)
     error_message = Column(Text, nullable=True)
@@ -120,4 +127,5 @@ class GeminiJob(Base):
     completed_at = Column(DateTime(timezone=True), nullable=True)
 
     def __repr__(self):
-        return f"<GeminiJob(job_id={self.job_id}, type={self.job_type}, status={self.status})>"
+        return f"<GeminiJob(job_id={self.job_id}, type={self.job_type}, status={self.status}, priority={self.priority})>"
+

routers/gemini.py

@@ -74,14 +74,18 @@ async def create_job(
     job_type: str,
     input_data: dict
 ) -> GeminiJob:
-    """Create a new job in the queue."""
+    """Create a new job in the queue with auto-assigned priority."""
+    from services.job_worker import get_priority_for_job_type
+    
     job_id = f"job_{uuid.uuid4().hex[:16]}"
+    priority = get_priority_for_job_type(job_type)
     
     job = GeminiJob(
         job_id=job_id,
         user_id=user.user_id,
         job_type=job_type,
         status="queued",
+        priority=priority,
         input_data=input_data
     )
     db.add(job)
@@ -344,6 +348,48 @@ async def download_video(
     )
 
 
+@router.post("/job/{job_id}/cancel")
+async def cancel_job(
+    job_id: str,
+    user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """
+    Cancel a queued job.
+    Only jobs with status 'queued' can be cancelled.
+    Processing/completed/failed jobs cannot be cancelled.
+    """
+    query = select(GeminiJob).where(
+        GeminiJob.job_id == job_id,
+        GeminiJob.user_id == user.user_id
+    )
+    result = await db.execute(query)
+    job = result.scalar_one_or_none()
+    
+    if not job:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Job not found"
+        )
+    
+    if job.status != "queued":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Cannot cancel job with status '{job.status}'. Only queued jobs can be cancelled."
+        )
+    
+    job.status = "cancelled"
+    job.completed_at = datetime.utcnow()
+    await db.commit()
+    
+    return {
+        "success": True,
+        "job_id": job.job_id,
+        "status": "cancelled",
+        "message": "Job cancelled successfully"
+    }
+
+
 @router.get("/models")
 async def get_models():
     """

services/job_worker.py

@@ -1,184 +1,219 @@
 """
-Background worker for processing Gemini jobs.
-Runs continuously, picks up queued jobs, processes them.
+Priority-Tier Worker Pool for processing Gemini jobs.
+
+Architecture:
+- 15 workers total: 5 fast (5s), 5 medium (30s), 5 slow (60s)
+- Workers pick jobs based on priority tier
+- Jobs are rescheduled with next_process_at if third-party not done
+- No blocking on third-party polling
 """
 import asyncio
 import logging
 import os
-from datetime import datetime
-from sqlalchemy import select
+from datetime import datetime, timedelta
+from typing import Optional, List
+from sqlalchemy import select, or_, and_
 from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
 
 from core.database import DATABASE_URL
 from core.models import GeminiJob
-from services.gemini_service import GeminiService, DOWNLOADS_DIR
+from services.gemini_service import GeminiService
 
 logger = logging.getLogger(__name__)
 
 # Worker configuration
-WORKER_POLL_INTERVAL = int(os.getenv("WORKER_POLL_INTERVAL", "5"))  # seconds
-MAX_CONCURRENT_VIDEO_JOBS = int(os.getenv("MAX_CONCURRENT_VIDEO_JOBS", "2"))
-MAX_CONCURRENT_IMAGE_JOBS = int(os.getenv("MAX_CONCURRENT_IMAGE_JOBS", "3"))
-MAX_CONCURRENT_TEXT_JOBS = int(os.getenv("MAX_CONCURRENT_TEXT_JOBS", "5"))
+FAST_WORKERS = int(os.getenv("FAST_WORKERS", "5"))
+MEDIUM_WORKERS = int(os.getenv("MEDIUM_WORKERS", "5"))
+SLOW_WORKERS = int(os.getenv("SLOW_WORKERS", "5"))
+
+FAST_INTERVAL = int(os.getenv("FAST_INTERVAL", "5"))      # 5 seconds
+MEDIUM_INTERVAL = int(os.getenv("MEDIUM_INTERVAL", "30"))  # 30 seconds
+SLOW_INTERVAL = int(os.getenv("SLOW_INTERVAL", "60"))      # 60 seconds
 
-# Track running jobs
-_running_jobs = {
-    "video": 0,
-    "image": 0,
-    "text": 0,
-    "analyze": 0,
-    "animation_prompt": 0
+# Job type to priority mapping
+JOB_PRIORITY_MAP = {
+    "text": "fast",
+    "analyze": "fast",
+    "animation_prompt": "fast",
+    "image": "medium",
+    "edit_image": "medium",
+    "video": "slow"
 }
-_lock = asyncio.Lock()
+
+def get_priority_for_job_type(job_type: str) -> str:
+    """Get the priority tier for a job type."""
+    return JOB_PRIORITY_MAP.get(job_type, "fast")
+
+def get_interval_for_priority(priority: str) -> int:
+    """Get the polling interval in seconds for a priority tier."""
+    if priority == "fast":
+        return FAST_INTERVAL
+    elif priority == "medium":
+        return MEDIUM_INTERVAL
+    else:
+        return SLOW_INTERVAL
 
 
-class JobWorker:
-    """Background worker for processing Gemini jobs."""
+class PriorityWorker:
+    """Worker that processes jobs of a specific priority tier."""
     
-    def __init__(self):
-        self.engine = create_async_engine(DATABASE_URL, echo=False)
-        self.async_session = async_sessionmaker(
-            self.engine,
-            class_=AsyncSession,
-            expire_on_commit=False
-        )
+    def __init__(self, worker_id: int, priority: str, poll_interval: int, session_maker):
+        self.worker_id = worker_id
+        self.priority = priority
+        self.poll_interval = poll_interval
+        self.session_maker = session_maker
         self._running = False
-        self._tasks = []
+        self._current_job_id: Optional[str] = None
     
     async def start(self):
-        """Start the worker."""
+        """Start the worker polling loop."""
         self._running = True
-        logger.info("Job worker started")
+        logger.info(f"Worker {self.worker_id} ({self.priority}) started, polling every {self.poll_interval}s")
         asyncio.create_task(self._poll_loop())
     
     async def stop(self):
         """Stop the worker."""
         self._running = False
-        # Wait for running tasks to complete
-        if self._tasks:
-            await asyncio.gather(*self._tasks, return_exceptions=True)
-        logger.info("Job worker stopped")
+        logger.info(f"Worker {self.worker_id} ({self.priority}) stopped")
     
     async def _poll_loop(self):
         """Main polling loop."""
         while self._running:
             try:
-                await self._process_queued_jobs()
+                await self._process_one_job()
             except Exception as e:
-                logger.error(f"Error in worker poll loop: {e}")
-            await asyncio.sleep(WORKER_POLL_INTERVAL)
+                logger.error(f"Worker {self.worker_id}: Error in poll loop: {e}")
+            await asyncio.sleep(self.poll_interval)
     
-    async def _process_queued_jobs(self):
-        """Find and process queued jobs."""
-        async with self.async_session() as session:
-            # Get queued jobs
+    async def _process_one_job(self):
+        """Find and process one job."""
+        async with self.session_maker() as session:
+            # Find a job to process
+            now = datetime.utcnow()
             query = select(GeminiJob).where(
-                GeminiJob.status == "queued"
-            ).order_by(GeminiJob.created_at)
+                and_(
+                    GeminiJob.priority == self.priority,
+                    GeminiJob.status.in_(["queued", "processing"]),
+                    or_(
+                        GeminiJob.next_process_at.is_(None),
+                        GeminiJob.next_process_at <= now
+                    )
+                )
+            ).order_by(GeminiJob.created_at).limit(1)
             
             result = await session.execute(query)
-            jobs = result.scalars().all()
+            job = result.scalar_one_or_none()
+            
+            if not job:
+                return  # No jobs to process
+            
+            self._current_job_id = job.job_id
             
-            for job in jobs:
-                # Check if we can process this job type
-                if not await self._can_process(job.job_type):
-                    continue
-                
-                # Mark as processing
-                async with _lock:
-                    _running_jobs[job.job_type] = _running_jobs.get(job.job_type, 0) + 1
-                
-                try:
-                    job.status = "processing"
-                    job.started_at = datetime.utcnow()
-                    await session.commit()
-                    
-                    # Process job in background
-                    task = asyncio.create_task(self._process_job(job.job_id))
-                    self._tasks.append(task)
-                    task.add_done_callback(lambda t: self._tasks.remove(t) if t in self._tasks else None)
-                except Exception as e:
-                    logger.error(f"Error starting job {job.job_id}: {e}")
-                    async with _lock:
-                        _running_jobs[job.job_type] = max(0, _running_jobs.get(job.job_type, 0) - 1)
-    
-    async def _can_process(self, job_type: str) -> bool:
-        """Check if we can process another job of this type."""
-        async with _lock:
-            current = _running_jobs.get(job_type, 0)
-            if job_type == "video":
-                return current < MAX_CONCURRENT_VIDEO_JOBS
-            elif job_type in ("image", "edit_image"):
-                return current < MAX_CONCURRENT_IMAGE_JOBS
-            else:  # text, analyze, animation_prompt
-                return current < MAX_CONCURRENT_TEXT_JOBS
-    
-    async def _process_job(self, job_id: str):
-        """Process a single job."""
-        async with self.async_session() as session:
             try:
-                # Get the job
-                query = select(GeminiJob).where(GeminiJob.job_id == job_id)
-                result = await session.execute(query)
-                job = result.scalar_one_or_none()
-                
-                if not job:
-                    logger.error(f"Job {job_id} not found")
-                    return
-                
-                logger.info(f"Processing job {job_id} (type: {job.job_type})")
-                
-                service = GeminiService()
-                input_data = job.input_data or {}
-                
-                if job.job_type == "video":
-                    await self._process_video_job(session, job, service, input_data)
-                elif job.job_type == "image":
-                    await self._process_image_job(session, job, service, input_data)
-                elif job.job_type == "text":
-                    await self._process_text_job(session, job, service, input_data)
-                elif job.job_type == "analyze":
-                    await self._process_analyze_job(session, job, service, input_data)
-                elif job.job_type == "animation_prompt":
-                    await self._process_animation_prompt_job(session, job, service, input_data)
-                else:
-                    job.status = "failed"
-                    job.error_message = f"Unknown job type: {job.job_type}"
-                    job.completed_at = datetime.utcnow()
-                    await session.commit()
-                
+                await self._process_job(session, job)
             except Exception as e:
-                logger.error(f"Error processing job {job_id}: {e}")
-                try:
-                    job.status = "failed"
-                    job.error_message = str(e)
-                    job.completed_at = datetime.utcnow()
-                    await session.commit()
-                except:
-                    pass
+                logger.error(f"Worker {self.worker_id}: Error processing job {job.job_id}: {e}")
+                job.status = "failed"
+                job.error_message = str(e)
+                job.completed_at = datetime.utcnow()
+                await session.commit()
             finally:
-                async with _lock:
-                    job_type = job.job_type if job else "unknown"
-                    _running_jobs[job_type] = max(0, _running_jobs.get(job_type, 0) - 1)
+                self._current_job_id = None
     
-    async def _process_video_job(self, session: AsyncSession, job: GeminiJob, service: GeminiService, input_data: dict):
-        """Process a video generation job."""
-        # Start video generation
-        result = await service.start_video_generation(
-            base64_image=input_data.get("base64_image", ""),
-            mime_type=input_data.get("mime_type", "image/jpeg"),
-            prompt=input_data.get("prompt", ""),
-            aspect_ratio=input_data.get("aspect_ratio", "16:9"),
-            resolution=input_data.get("resolution", "720p"),
-            number_of_videos=input_data.get("number_of_videos", 1)
-        )
+    async def _process_job(self, session: AsyncSession, job: GeminiJob):
+        """Process a single job."""
+        logger.info(f"Worker {self.worker_id}: Processing job {job.job_id} (type: {job.job_type}, status: {job.status})")
+        
+        service = GeminiService()
+        input_data = job.input_data or {}
         
-        # Save third party ID
-        job.third_party_id = result.get("gemini_operation_name")
-        await session.commit()
+        # If queued, start the operation
+        if job.status == "queued":
+            job.status = "processing"
+            job.started_at = datetime.utcnow()
+            await session.commit()
+            
+            # Start the third-party operation
+            await self._start_operation(session, job, service, input_data)
         
-        # Poll until done
-        while True:
+        # Check status for operations that need polling (video)
+        if job.job_type == "video" and job.third_party_id:
+            await self._check_video_status(session, job, service)
+        # For synchronous operations (already completed in _start_operation)
+        # Nothing more to do
+    
+    async def _start_operation(self, session: AsyncSession, job: GeminiJob, service: GeminiService, input_data: dict):
+        """Start the third-party operation based on job type."""
+        if job.job_type == "video":
+            # Start async video generation
+            result = await service.start_video_generation(
+                base64_image=input_data.get("base64_image", ""),
+                mime_type=input_data.get("mime_type", "image/jpeg"),
+                prompt=input_data.get("prompt", ""),
+                aspect_ratio=input_data.get("aspect_ratio", "16:9"),
+                resolution=input_data.get("resolution", "720p"),
+                number_of_videos=input_data.get("number_of_videos", 1)
+            )
+            job.third_party_id = result.get("gemini_operation_name")
+            # Schedule first status check
+            job.next_process_at = datetime.utcnow() + timedelta(seconds=self.poll_interval)
+            await session.commit()
+            
+        elif job.job_type == "image":
+            # Synchronous image edit
+            result = await service.edit_image(
+                base64_image=input_data.get("base64_image", ""),
+                mime_type=input_data.get("mime_type", "image/jpeg"),
+                prompt=input_data.get("prompt", "")
+            )
+            job.status = "completed"
+            job.output_data = {"image": result}
+            job.completed_at = datetime.utcnow()
+            await session.commit()
+            
+        elif job.job_type == "text":
+            # Synchronous text generation
+            result = await service.generate_text(
+                prompt=input_data.get("prompt", ""),
+                model=input_data.get("model")
+            )
+            job.status = "completed"
+            job.output_data = {"text": result}
+            job.completed_at = datetime.utcnow()
+            await session.commit()
+            
+        elif job.job_type == "analyze":
+            # Synchronous image analysis
+            result = await service.analyze_image(
+                base64_image=input_data.get("base64_image", ""),
+                mime_type=input_data.get("mime_type", "image/jpeg"),
+                prompt=input_data.get("prompt", "")
+            )
+            job.status = "completed"
+            job.output_data = {"analysis": result}
+            job.completed_at = datetime.utcnow()
+            await session.commit()
+            
+        elif job.job_type == "animation_prompt":
+            # Synchronous animation prompt generation
+            result = await service.generate_animation_prompt(
+                base64_image=input_data.get("base64_image", ""),
+                mime_type=input_data.get("mime_type", "image/jpeg"),
+                custom_prompt=input_data.get("custom_prompt")
+            )
+            job.status = "completed"
+            job.output_data = {"prompt": result}
+            job.completed_at = datetime.utcnow()
+            await session.commit()
+        else:
+            job.status = "failed"
+            job.error_message = f"Unknown job type: {job.job_type}"
+            job.completed_at = datetime.utcnow()
+            await session.commit()
+    
+    async def _check_video_status(self, session: AsyncSession, job: GeminiJob, service: GeminiService):
+        """Check video generation status and reschedule if not done."""
+        try:
             status_result = await service.check_video_status(job.third_party_id)
             
             if status_result.get("done"):
@@ -197,82 +232,93 @@ class JobWorker:
                     job.error_message = status_result.get("error", "Unknown error")
                 
                 job.completed_at = datetime.utcnow()
-                await session.commit()
-                break
+            else:
+                # Not done - reschedule
+                job.retry_count += 1
+                job.next_process_at = datetime.utcnow() + timedelta(seconds=self.poll_interval)
+                logger.debug(f"Job {job.job_id}: Not done, retry #{job.retry_count}, next check at {job.next_process_at}")
             
-            await asyncio.sleep(10)  # Poll every 10 seconds
+            await session.commit()
+            
+        except Exception as e:
+            logger.error(f"Error checking video status for {job.job_id}: {e}")
+            job.retry_count += 1
+            job.next_process_at = datetime.utcnow() + timedelta(seconds=self.poll_interval)
+            if job.retry_count > 60:  # ~1 hour of retries
+                job.status = "failed"
+                job.error_message = f"Max retries exceeded: {str(e)}"
+                job.completed_at = datetime.utcnow()
+            await session.commit()
+
+
+class WorkerPool:
+    """Pool of priority-tier workers."""
     
-    async def _process_image_job(self, session: AsyncSession, job: GeminiJob, service: GeminiService, input_data: dict):
-        """Process an image edit job."""
-        result = await service.edit_image(
-            base64_image=input_data.get("base64_image", ""),
-            mime_type=input_data.get("mime_type", "image/jpeg"),
-            prompt=input_data.get("prompt", "")
+    def __init__(self):
+        self.engine = create_async_engine(DATABASE_URL, echo=False)
+        self.session_maker = async_sessionmaker(
+            self.engine,
+            class_=AsyncSession,
+            expire_on_commit=False
         )
-        
-        job.status = "completed"
-        job.output_data = {"image": result}
-        job.completed_at = datetime.utcnow()
-        await session.commit()
+        self.workers: List[PriorityWorker] = []
+        self._running = False
     
-    async def _process_text_job(self, session: AsyncSession, job: GeminiJob, service: GeminiService, input_data: dict):
-        """Process a text generation job."""
-        result = await service.generate_text(
-            prompt=input_data.get("prompt", ""),
-            model=input_data.get("model")
-        )
+    async def start(self):
+        """Start all workers."""
+        self._running = True
+        worker_id = 0
         
-        job.status = "completed"
-        job.output_data = {"text": result}
-        job.completed_at = datetime.utcnow()
-        await session.commit()
-    
-    async def _process_analyze_job(self, session: AsyncSession, job: GeminiJob, service: GeminiService, input_data: dict):
-        """Process an image analysis job."""
-        result = await service.analyze_image(
-            base64_image=input_data.get("base64_image", ""),
-            mime_type=input_data.get("mime_type", "image/jpeg"),
-            prompt=input_data.get("prompt", "")
-        )
+        # Create fast workers
+        for i in range(FAST_WORKERS):
+            worker = PriorityWorker(worker_id, "fast", FAST_INTERVAL, self.session_maker)
+            self.workers.append(worker)
+            await worker.start()
+            worker_id += 1
         
-        job.status = "completed"
-        job.output_data = {"analysis": result}
-        job.completed_at = datetime.utcnow()
-        await session.commit()
-    
-    async def _process_animation_prompt_job(self, session: AsyncSession, job: GeminiJob, service: GeminiService, input_data: dict):
-        """Process an animation prompt generation job."""
-        result = await service.generate_animation_prompt(
-            base64_image=input_data.get("base64_image", ""),
-            mime_type=input_data.get("mime_type", "image/jpeg"),
-            custom_prompt=input_data.get("custom_prompt")
-        )
+        # Create medium workers
+        for i in range(MEDIUM_WORKERS):
+            worker = PriorityWorker(worker_id, "medium", MEDIUM_INTERVAL, self.session_maker)
+            self.workers.append(worker)
+            await worker.start()
+            worker_id += 1
+        
+        # Create slow workers
+        for i in range(SLOW_WORKERS):
+            worker = PriorityWorker(worker_id, "slow", SLOW_INTERVAL, self.session_maker)
+            self.workers.append(worker)
+            await worker.start()
+            worker_id += 1
         
-        job.status = "completed"
-        job.output_data = {"prompt": result}
-        job.completed_at = datetime.utcnow()
-        await session.commit()
+        logger.info(f"WorkerPool started with {len(self.workers)} workers: {FAST_WORKERS} fast, {MEDIUM_WORKERS} medium, {SLOW_WORKERS} slow")
+    
+    async def stop(self):
+        """Stop all workers."""
+        self._running = False
+        for worker in self.workers:
+            await worker.stop()
+        logger.info("WorkerPool stopped")
 
 
-# Singleton worker instance
-_worker: JobWorker = None
+# Singleton pool instance
+_pool: Optional[WorkerPool] = None
 
 
-def get_worker() -> JobWorker:
-    """Get the global worker instance."""
-    global _worker
-    if _worker is None:
-        _worker = JobWorker()
-    return _worker
+def get_pool() -> WorkerPool:
+    """Get the global worker pool instance."""
+    global _pool
+    if _pool is None:
+        _pool = WorkerPool()
+    return _pool
 
 
 async def start_worker():
-    """Start the background worker."""
-    worker = get_worker()
-    await worker.start()
+    """Start the background worker pool."""
+    pool = get_pool()
+    await pool.start()
 
 
 async def stop_worker():
-    """Stop the background worker."""
-    worker = get_worker()
-    await worker.stop()
+    """Stop the background worker pool."""
+    pool = get_pool()
+    await pool.stop()

tests/test_worker_pool.py

@@ -0,0 +1,147 @@
+"""
+Tests for Priority-Tier Worker Pool implementation.
+Tests job creation with priority, cancel endpoint, and worker assignment.
+"""
+import pytest
+import asyncio
+from unittest.mock import patch, MagicMock, AsyncMock
+from datetime import datetime, timedelta
+
+# Test the priority mapping
+from services.job_worker import (
+    get_priority_for_job_type,
+    get_interval_for_priority,
+    PriorityWorker,
+    WorkerPool,
+    JOB_PRIORITY_MAP
+)
+
+
+class TestPriorityMapping:
+    """Test job type to priority mapping."""
+    
+    def test_text_job_is_fast(self):
+        assert get_priority_for_job_type("text") == "fast"
+    
+    def test_analyze_job_is_fast(self):
+        assert get_priority_for_job_type("analyze") == "fast"
+    
+    def test_animation_prompt_is_fast(self):
+        assert get_priority_for_job_type("animation_prompt") == "fast"
+    
+    def test_image_job_is_medium(self):
+        assert get_priority_for_job_type("image") == "medium"
+    
+    def test_edit_image_is_medium(self):
+        assert get_priority_for_job_type("edit_image") == "medium"
+    
+    def test_video_job_is_slow(self):
+        assert get_priority_for_job_type("video") == "slow"
+    
+    def test_unknown_job_defaults_to_fast(self):
+        assert get_priority_for_job_type("unknown_type") == "fast"
+
+
+class TestIntervalMapping:
+    """Test priority to interval mapping."""
+    
+    def test_fast_interval(self):
+        assert get_interval_for_priority("fast") == 5
+    
+    def test_medium_interval(self):
+        assert get_interval_for_priority("medium") == 30
+    
+    def test_slow_interval(self):
+        assert get_interval_for_priority("slow") == 60
+    
+    def test_unknown_defaults_to_slow(self):
+        assert get_interval_for_priority("unknown") == 60
+
+
+class TestJobPriorityMap:
+    """Test that all expected job types are covered."""
+    
+    def test_all_job_types_have_priority(self):
+        expected_types = ["text", "analyze", "animation_prompt", "image", "edit_image", "video"]
+        for job_type in expected_types:
+            assert job_type in JOB_PRIORITY_MAP, f"Job type '{job_type}' not in priority map"
+
+
+class TestWorkerPoolConfiguration:
+    """Test worker pool creates correct number of workers."""
+    
+    @pytest.mark.asyncio
+    async def test_creates_15_workers(self):
+        """Test that WorkerPool creates 15 workers (5 fast, 5 medium, 5 slow)."""
+        pool = WorkerPool()
+        
+        # Start pool (workers will be created)
+        # Mock to prevent actual polling
+        with patch.object(PriorityWorker, '_poll_loop', new_callable=AsyncMock):
+            await pool.start()
+        
+        assert len(pool.workers) == 15
+        
+        # Count by priority
+        fast_workers = [w for w in pool.workers if w.priority == "fast"]
+        medium_workers = [w for w in pool.workers if w.priority == "medium"]
+        slow_workers = [w for w in pool.workers if w.priority == "slow"]
+        
+        assert len(fast_workers) == 5
+        assert len(medium_workers) == 5
+        assert len(slow_workers) == 5
+        
+        await pool.stop()
+    
+    @pytest.mark.asyncio
+    async def test_workers_have_correct_intervals(self):
+        """Test that workers have correct poll intervals."""
+        pool = WorkerPool()
+        
+        with patch.object(PriorityWorker, '_poll_loop', new_callable=AsyncMock):
+            await pool.start()
+        
+        for worker in pool.workers:
+            if worker.priority == "fast":
+                assert worker.poll_interval == 5
+            elif worker.priority == "medium":
+                assert worker.poll_interval == 30
+            elif worker.priority == "slow":
+                assert worker.poll_interval == 60
+        
+        await pool.stop()
+
+
+class TestPriorityWorker:
+    """Test individual worker behavior."""
+    
+    def test_worker_has_correct_attributes(self):
+        """Test worker initialization."""
+        worker = PriorityWorker(0, "fast", 5, None)
+        
+        assert worker.worker_id == 0
+        assert worker.priority == "fast"
+        assert worker.poll_interval == 5
+        assert worker._running == False
+        assert worker._current_job_id is None
+
+
+# Integration test with the actual router
+class TestCancelEndpoint:
+    """Test cancel job endpoint logic."""
+    
+    def test_only_queued_jobs_can_be_cancelled(self):
+        """Verify the cancellation logic - only queued status allowed."""
+        valid_statuses = ["queued"]
+        invalid_statuses = ["processing", "completed", "failed", "cancelled"]
+        
+        # This is a logic validation, actual HTTP testing would need the app
+        for status in valid_statuses:
+            assert status == "queued"
+        
+        for status in invalid_statuses:
+            assert status != "queued"
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])