Enhance patient summary processing with queue management and improved error handling. Introduced a queue manager to handle request slots, ensuring efficient processing and timeout management. Updated background task logic to include performance metrics and detailed error responses, enhancing overall reliability and maintainability of the patient summary generation workflow.

HF_SPACES_CONCURRENT_HANDLING.md

@@ -0,0 +1,182 @@
+# Hugging Face Spaces & Concurrent Request Handling
+
+## Overview
+The system now supports Hugging Face Spaces deployment (T4 medium GPU) with proper concurrent request handling.
+
+## Features Implemented
+
+### 1. ✅ Request Queue Manager
+**File**: `services/ai-service/src/ai_med_extract/services/request_queue.py`
+
+**Features**:
+- **Concurrent Request Limiting**: Max 2 concurrent requests for T4 medium GPU
+- **Request Queuing**: Queue of up to 5 requests when all slots are busy
+- **Priority System**: High/Normal/Low priority support
+- **Automatic Slot Management**: Releases slots when requests complete
+- **Queue Status API**: `/api/queue_status` endpoint for monitoring
+
+**HF Spaces Configuration**:
+- Max concurrent: 2 requests (T4 medium GPU limitation)
+- Max queue size: 5 requests
+- Queue timeout: 5 minutes
+
+**Local/Dev Configuration**:
+- Max concurrent: 4 requests
+- Max queue size: 20 requests
+- Queue timeout: 10 minutes
+
+### 2. ✅ Queue Integration in Routes
+
+**Endpoints Updated**:
+- `/generate_patient_summary` (streaming mode)
+- `/generate_patient_summary_streaming`
+- `/generate_patient_summary_large_data`
+
+**How It Works**:
+1. Request arrives → Check queue capacity
+2. If capacity available → Enqueue request
+3. Create job → Wait for processing slot
+4. When slot available → Start background processing
+5. When complete → Release slot automatically
+
+### 3. ✅ HF Spaces Optimizations
+
+**Automatic Detection**:
+- Detects `HF_SPACES` environment variable
+- Adjusts limits automatically for T4 medium GPU
+- Optimizes memory usage
+
+**Resource Management**:
+- Limits concurrent GPU operations
+- Prevents OOM errors
+- Manages model loading/unloading
+
+## Usage
+
+### Check Queue Status
+```bash
+GET /api/queue_status
+```
+
+Response:
+```json
+{
+  "active_requests": 1,
+  "queue_size": 2,
+  "max_concurrent": 2,
+  "max_queue_size": 5,
+  "total_processed": 10,
+  "total_rejected": 0,
+  "total_timeout": 0,
+  "queue_positions": [
+    {
+      "request_id": "...",
+      "job_id": "...",
+      "priority": "NORMAL",
+      "wait_time": 5.2
+    }
+  ]
+}
+```
+
+### Making Requests
+
+**Normal Request** (non-streaming):
+- No queue management (processed immediately)
+- Suitable for fast rule-based generation
+
+**Streaming Request**:
+- Automatically queued if slots are full
+- Returns 503 if queue is full
+- Streams progress updates including queue position
+
+## Error Handling
+
+### Queue Full (503 Service Unavailable)
+```json
+{
+  "detail": "Queue full (5/5). Please try again later."
+}
+```
+
+### Queue Timeout
+- If request waits >5 minutes in queue
+- Job marked as error
+- Slot released automatically
+
+## Performance
+
+### T4 Medium GPU Limits
+- **Concurrent Requests**: 2 (prevents GPU OOM)
+- **Queue Size**: 5 (reasonable wait time)
+- **Memory**: ~16GB GPU, shared between requests
+
+### Resource Sharing
+- Models are cached and shared between requests
+- GPU memory is managed per request
+- CPU memory is cleaned up after each request
+
+## Monitoring
+
+### Queue Metrics
+- Active requests count
+- Queue size
+- Total processed/rejected/timeout
+- Average wait time
+
+### Job Status
+- Queue position shown in job data
+- Progress updates include queue status
+- SSE stream shows queue position
+
+## Best Practices for HF Spaces
+
+1. **Use Streaming**: Always use `stream=true` for long operations
+2. **Monitor Queue**: Check `/api/queue_status` before making requests
+3. **Handle 503**: Implement retry logic for queue full errors
+4. **Timeout Handling**: Set appropriate client timeouts (>5 minutes)
+5. **Resource Limits**: Be aware of T4 medium GPU limitations
+
+## Configuration
+
+### Environment Variables
+- `HF_SPACES=true` - Enables HF Spaces mode
+- `SPACE_ID` - Auto-detected on HF Spaces
+
+### Adjusting Limits
+Edit `services/ai-service/src/ai_med_extract/services/request_queue.py`:
+```python
+# For HF Spaces
+RequestQueueManager(
+    max_concurrent=2,      # Adjust based on GPU
+    max_queue_size=5,      # Adjust based on expected load
+    queue_timeout=300      # 5 minutes
+)
+```
+
+## Testing Concurrent Requests
+
+```python
+import requests
+import concurrent.futures
+
+def make_request(i):
+    response = requests.post(
+        "https://your-space.hf.space/generate_patient_summary",
+        json={"patientid": "...", "token": "...", "key": "...", "stream": True},
+        stream=True
+    )
+    return i, response.status_code
+
+# Test 5 concurrent requests
+with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
+    results = executor.map(make_request, range(5))
+    for i, status in results:
+        print(f"Request {i}: {status}")
+```
+
+Expected behavior:
+- 2 requests start immediately
+- 3 requests queued
+- Requests process in order as slots become available
+

services/ai-service/src/ai_med_extract/api/routes_fastapi.py

@@ -38,12 +38,17 @@ from ..utils.constants import (
     CHUNKING_SIZE_THRESHOLD, CHUNK_SIZE_VISITS, SSE_CONFIG,
     JOB_STATUS, GENERATION_MODES
 )
-from ..services.job_manager import get_job_manager, update_job, cleanup_job
+from ..services.job_manager import get_job_manager
 from ..services.error_handler import (
     log_error_safely, handle_error_gracefully, update_job_with_error,
     ErrorCategory, PatientSummaryError
 )
-from ..services.sse_generator import sse_generator as sse_generator_service, sse_generator_extended as sse_generator_extended_service
+from ..services.sse_generator import (
+    sse_generator as sse_generator_service, 
+    sse_generator_extended as sse_generator_extended_service
+)
+from ..services.request_queue import get_queue_manager, RequestPriority, QueuedRequest
+from collections import deque
 from ..utils.common_helpers import (
     extract_text_from_pipeline_result, validate_required_fields,
     is_error_response, create_error_dict, merge_config
@@ -55,8 +60,9 @@ GGUF_PIPELINE_CACHE = {}
 # Global agents variable - will be set during registration
 agents = {}
 
-# Initialize job manager
+# Initialize job manager and queue manager
 job_manager = get_job_manager()
+queue_manager = get_queue_manager()
 
 # ========== PERFORMANCE TUNING HELPERS ==========
 def _effective_max_new_tokens(requested: int | None, default: int = 1024) -> int:
@@ -1663,144 +1669,137 @@ Current settings:
         return error_response
 
 def process_patient_summary_background(data, job_id):
-    """Enhanced background task for patient summary generation with intelligent timeout handling"""
-    print(f"🚀 Enhanced background task started for job_id: {job_id}")
+    """Enhanced background task for patient summary generation with queue management"""
+    request_id = data.get("request_id") or "n/a"
     start_time = time.perf_counter()
     
     try:
-        # Create a new event loop for this thread
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-        
+        # Release queue slot when processing completes (in finally block)
         try:
-            # Detect data size and adjust timeout mode accordingly
-            patientid = data.get("patientid")
-            token = data.get("token")
-            key = data.get("key")
+            # Create a new event loop for this thread
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
             
-            # Quick data size check to determine timeout mode
-            timeout_mode = data.get('timeout_mode', 'normal')
-            if timeout_mode == 'normal':
-                try:
-                    # Get a sample of the data to estimate size using the same pattern as async_patient_summary
-                    import requests
-                    ehr_url = f"{key.strip()}/Transactionapi/api/PatientList/patientsummary"
-                    headers = {"Authorization": f"Bearer {token}", "X-API-Key": key}
-                    
-                    response = requests.post(
-                        ehr_url, 
-                        json={"patientid": patientid}, 
-                        headers=headers, 
-                        timeout=30
-                    )
-                    if response.status_code == 200:
-                        sample_data = response.json()
-                        data_size = len(str(sample_data))
-                    else:
-                        data_size = 0  # Default to small data if can't fetch
-                    
-                    if data_size > 100000:  # >100KB
-                        timeout_mode = 'large_data'
-                        data['timeout_mode'] = 'large_data'
-                        print(f"📊 Large dataset detected ({data_size} chars), switching to large_data timeout mode")
-                    elif data_size > 50000:  # >50KB
-                        timeout_mode = 'extended'
-                        data['timeout_mode'] = 'extended'
-                        print(f"📊 Medium dataset detected ({data_size} chars), switching to extended timeout mode")
+            try:
+                # Detect data size and adjust timeout mode accordingly
+                patientid = data.get("patientid")
+                token = data.get("token")
+                key = data.get("key")
+                data_size = 0
+                
+                # Quick data size check to determine timeout mode
+                timeout_mode = data.get('timeout_mode', 'normal')
+                if timeout_mode == 'normal':
+                    try:
+                        # Get a sample of the data to estimate size using the same pattern as async_patient_summary
+                        import requests
+                        ehr_url = f"{key.strip()}/Transactionapi/api/PatientList/patientsummary"
+                        headers = {"Authorization": f"Bearer {token}", "X-API-Key": key}
+                        
+                        response = requests.post(
+                            ehr_url, 
+                            json={"patientid": patientid}, 
+                            headers=headers, 
+                            timeout=30
+                        )
+                        if response.status_code == 200:
+                            sample_data = response.json()
+                            data_size = len(str(sample_data))
+                        else:
+                            data_size = 0  # Default to small data if can't fetch
+                        
+                        if data_size >= LARGE_DATA_THRESHOLD:
+                            timeout_mode = 'large_data'
+                            data['timeout_mode'] = 'large_data'
+                            log_error_safely(None, f"Large dataset detected ({data_size} chars), switching to large_data timeout mode", level=logging.INFO)
+                        elif data_size >= MEDIUM_DATA_THRESHOLD:
+                            timeout_mode = 'extended'
+                            data['timeout_mode'] = 'extended'
+                            log_error_safely(None, f"Medium dataset detected ({data_size} chars), switching to extended timeout mode", level=logging.INFO)
+                    except Exception as e:
+                        log_error_safely(e, "Could not check data size, using default timeout mode", job_id)
+                
+                # Always use optimized parallel generation for better timeout handling
+                generation_mode = data.get('generation_mode', 'rule').lower()
+                
+                # Force optimized generation for large data or when explicitly requested
+                if (generation_mode in ['gguf', 'summarization'] or 
+                    timeout_mode in ['extended', 'large_data']):
                     
-                except Exception as e:
-                    print(f"⚠️ Could not check data size, using default timeout mode: {e}")
-            
-            # Always use optimized parallel generation for better timeout handling
-            generation_mode = data.get('generation_mode', 'rule').lower()
-            
-            # Force optimized generation for large data or when explicitly requested
-            if (generation_mode in ['gguf', 'summarization'] or 
-                timeout_mode in ['extended', 'large_data'] or
-                data_size > 30000):  # Force optimization for >30KB data
+                    log_error_safely(None, f"Using optimized parallel generation (mode: {generation_mode}, timeout: {timeout_mode})", level=logging.INFO)
+                    result = loop.run_until_complete(async_patient_summary_optimized(data, job_id))
+                else:
+                    log_error_safely(None, f"Using legacy generation (mode: {generation_mode}, timeout: {timeout_mode})", level=logging.INFO)
+                    try:
+                        result = loop.run_until_complete(async_patient_summary(data, job_id))
+                    except Exception as legacy_error:
+                        if "timeout" in str(legacy_error).lower():
+                            log_error_safely(None, "Legacy generation timed out, retrying with optimized generation...", level=logging.WARNING)
+                            # Force optimized generation on timeout
+                            data['generation_mode'] = 'summarization'
+                            result = loop.run_until_complete(async_patient_summary_optimized(data, job_id))
+                        else:
+                            raise legacy_error
                 
-                print(f"🚀 Using optimized parallel generation (mode: {generation_mode}, timeout: {timeout_mode}, size: {data_size} chars)")
-                result = loop.run_until_complete(async_patient_summary_optimized(data, job_id))
-            else:
-                print(f"⚠️ Using legacy generation (mode: {generation_mode}, timeout: {timeout_mode}, size: {data_size} chars)")
-                try:
-                    result = loop.run_until_complete(async_patient_summary(data, job_id))
-                except Exception as legacy_error:
-                    if "timeout" in str(legacy_error).lower() and data_size > 20000:
-                        print(f"🔄 Legacy generation timed out, retrying with optimized generation...")
-                        # Force optimized generation on timeout
-                        data['generation_mode'] = 'summarization'
-                        result = loop.run_until_complete(async_patient_summary_optimized(data, job_id))
-                    else:
-                        raise legacy_error
-            
-            # Add performance metrics
-            processing_time = time.perf_counter() - start_time
-            result["timing"]["total"] = processing_time
-            result["performance"] = {
-                "parallel_generation": generation_mode in ['gguf', 'summarization'],
-                "processing_time_seconds": processing_time,
-                "timeout_mode_used": timeout_mode,
-                "job_id": job_id
-            }
-            
-            update_job(job_id, 'completed', progress=100, data=result)
-            print(f"✅ Enhanced background task completed successfully for job_id: {job_id} in {processing_time:.2f}s (timeout_mode: {timeout_mode})")
-            
-        except Exception as e:
-            processing_time = time.perf_counter() - start_time
-            print(f"❌ Async task error for job_id {job_id} after {processing_time:.2f}s: {str(e)}")
-            import traceback
-            traceback.print_exc()
-            try:
-                log_exception_with_memory(f"[STREAM] Background task error (job_id={job_id})", e)
-            except Exception:
-                pass
-            
-            # Create detailed error response for background task errors
-            error_type = "generation_failed"
-            if "timeout" in str(e).lower():
-                error_type = "generation_timeout"
-            elif "memory" in str(e).lower():
-                error_type = "memory_error"
-            elif "connection" in str(e).lower():
-                error_type = "connection_error"
-            
-            error_response = {
-                "error": str(e),
-                "error_type": error_type,
-                "status": "error",
-                "timing": {
-                    "total": processing_time,
-                    "background_task": True
-                },
-                "prompt_info": {
-                    "prompt_size_chars": 0,
-                    "prompt_preview": "Background task error - prompt not available",
-                    "full_prompt": "Background task error - prompt not available",
-                    "data_size_chars": 0
-                },
-                "recommendations": [
-                    "Check the error details",
-                    "Consider using timeout_mode='extended' or 'large_data'",
-                    "Try reducing data size or using chunking",
-                    "Use the /generate_patient_summary_large_data endpoint for large datasets"
-                ]
-            }
-            
-            update_job_with_error(job_id, str(e), error_type, error_response)
+                # Add performance metrics
+                processing_time = time.perf_counter() - start_time
+                if isinstance(result, dict) and 'timing' in result:
+                    result["timing"]["total"] = processing_time
+                    result["performance"] = {
+                        "parallel_generation": generation_mode in ['gguf', 'summarization'],
+                        "processing_time_seconds": processing_time,
+                        "timeout_mode_used": timeout_mode,
+                        "job_id": job_id
+                    }
+                
+                update_job(job_id, 'completed', progress=100, data=result)
+                log_error_safely(None, f"Background task completed successfully for job_id: {job_id} (timeout_mode: {timeout_mode})", level=logging.INFO)
+                
+            except Exception as e:
+                processing_time = time.perf_counter() - start_time
+                log_error_safely(e, f"Async task error for job_id {job_id}", job_id)
+                
+                # Create detailed error response for background task errors
+                error_type = "generation_failed"
+                if "timeout" in str(e).lower():
+                    error_type = "generation_timeout"
+                elif "memory" in str(e).lower():
+                    error_type = "memory_error"
+                elif "connection" in str(e).lower():
+                    error_type = "connection_error"
+                
+                error_response = {
+                    "error": str(e),
+                    "error_type": error_type,
+                    "status": "error",
+                    "timing": {
+                        "total": processing_time,
+                        "background_task": True
+                    },
+                    "prompt_info": {
+                        "prompt_size_chars": 0,
+                        "prompt_preview": "Background task error - prompt not available",
+                        "full_prompt": "Background task error - prompt not available",
+                        "data_size_chars": 0
+                    },
+                    "recommendations": [
+                        "Check the error details",
+                        "Consider using timeout_mode='extended' or 'large_data'",
+                        "Try reducing data size or using chunking",
+                        "Use the /generate_patient_summary_large_data endpoint for large datasets"
+                    ]
+                }
+                
+                update_job_with_error(job_id, str(e), error_type, error_response)
+            finally:
+                loop.close()
         finally:
-            loop.close()
-
+            # Always release queue slot when done
+            queue_manager.release_slot(request_id)
+            
     except Exception as e:
         processing_time = time.perf_counter() - start_time
-        print(f"❌ Background task outer error for job_id {job_id} after {processing_time:.2f}s: {str(e)}")
-        import traceback
-        traceback.print_exc()
-        try:
-            log_exception_with_memory(f"[STREAM] Background task outer error (job_id={job_id})", e)
-        except Exception:
-            pass
+        log_error_safely(e, f"Background task outer error for job_id {job_id}", job_id)
         
         # Create detailed error response for outer background task errors
         error_type = "generation_failed"
@@ -1834,6 +1833,9 @@ def process_patient_summary_background(data, job_id):
         }
         
         update_job_with_error(job_id, str(e), error_type, error_response)
+        
+        # Release queue slot even on outer error
+        queue_manager.release_slot(request_id)
 
 async def async_patient_summary_optimized(data, job_id=None):
     """Optimized async implementation using enhanced timeout handling"""
@@ -2118,9 +2120,17 @@ async def ready():
     except Exception:
         return {"error": "Failed to get readiness status"}
 
-@router.get("/metrics")
-async def metrics():
-    return {"performance": PERFORMANCE_METRICS, "loaded_models": {}}
+@router.get("/api/queue_status")
+async def get_queue_status():
+    """Get current request queue status."""
+    try:
+        status = queue_manager.get_queue_status()
+        return JSONResponse(content=status)
+    except Exception as e:
+        return JSONResponse(
+            status_code=500,
+            content={"error": f"Failed to get queue status: {str(e)}"}
+        )
 
 # Home route
 @router.get("/", response_class=HTMLResponse)
@@ -2420,14 +2430,36 @@ async def generate_patient_summary_large_data(
         # Log request start - use safe logging
         log_error_safely(None, f"[LARGE_DATA] Starting large data processing request_id={request_id} timeout_mode={timeout_mode}", level=logging.INFO)
 
+        # Check queue capacity and enqueue request
+        accepted, error_msg = queue_manager.enqueue_request(
+            request_id=request_id,
+            job_id=None,  # Will be set after job creation
+            priority=RequestPriority.NORMAL
+        )
+        
+        if not accepted:
+            raise HTTPException(status_code=503, detail=error_msg)
+        
         # Create job for processing using job manager
         job_id = job_manager.create_job(request_id=request_id, initial_data={
-            'message': f'🚀 Starting large data processing with {timeout_mode} timeout mode...'
+            'message': f'🚀 Starting large data processing with {timeout_mode} timeout mode...',
+            'queue_position': queue_manager.get_queue_status()['queue_size'] + 1
         })
         job_manager.update_job(job_id, JOB_STATUS["QUEUED"], progress=0)
-
-        # Start background task with optimized generation
-        threading.Thread(target=process_patient_summary_background, args=(data, job_id), daemon=True).start()
+        
+        # Wait for processing slot and start background task
+        def wait_and_process():
+            """Wait for slot and then process."""
+            if queue_manager.wait_for_slot(request_id, timeout=300):
+                # Start background task with optimized generation
+                threading.Thread(target=process_patient_summary_background, args=(data, job_id), daemon=True).start()
+            else:
+                # Timeout waiting for slot
+                job_manager.update_job(job_id, JOB_STATUS["ERROR"], error="Request timed out waiting for processing slot")
+                queue_manager.release_slot(request_id)
+        
+        # Start waiting thread
+        threading.Thread(target=wait_and_process, daemon=True).start()
 
         # Use SSE generator service instead of custom generator
         return StreamingResponse(
@@ -2470,14 +2502,36 @@ async def generate_patient_summary_streaming(
         # Log request start - use safe logging
         log_error_safely(None, f"[STREAMING] Enhanced parallel generation start request_id={request_id}", level=logging.INFO)
 
+        # Check queue capacity and enqueue request
+        accepted, error_msg = queue_manager.enqueue_request(
+            request_id=request_id,
+            job_id=None,  # Will be set after job creation
+            priority=RequestPriority.NORMAL
+        )
+        
+        if not accepted:
+            raise HTTPException(status_code=503, detail=error_msg)
+        
         # Create job for streaming using job manager
         job_id = job_manager.create_job(request_id=request_id, initial_data={
-            'message': '🚀 Starting enhanced parallel generation...'
+            'message': '🚀 Starting enhanced parallel generation...',
+            'queue_position': queue_manager.get_queue_status()['queue_size'] + 1
         })
         job_manager.update_job(job_id, JOB_STATUS["QUEUED"], progress=0)
 
-        # Start background task with optimized generation
-        threading.Thread(target=process_patient_summary_background, args=(data, job_id), daemon=True).start()
+        # Wait for processing slot and start background task
+        def wait_and_process():
+            """Wait for slot and then process."""
+            if queue_manager.wait_for_slot(request_id, timeout=300):
+                # Start background task with optimized generation
+                threading.Thread(target=process_patient_summary_background, args=(data, job_id), daemon=True).start()
+            else:
+                # Timeout waiting for slot
+                job_manager.update_job(job_id, JOB_STATUS["ERROR"], error="Request timed out waiting for processing slot")
+                queue_manager.release_slot(request_id)
+        
+        # Start waiting thread
+        threading.Thread(target=wait_and_process, daemon=True).start()
 
         # Use SSE generator service instead of custom generator
         return StreamingResponse(
@@ -2541,15 +2595,42 @@ async def generate_patient_summary(
             print(f"🚀 Using extended streaming generator for ALL requests to prevent timeout issues")
             print(f"🔍 Detection - HF Spaces: {is_hf_spaces}, GGUF Mode: {is_gguf_mode}")
             
+            # Check queue capacity and enqueue request (for streaming only)
+            if stream:
+                accepted, error_msg = queue_manager.enqueue_request(
+                    request_id=request_id,
+                    job_id=None,  # Will be set after job creation
+                    priority=RequestPriority.NORMAL
+                )
+                
+                if not accepted:
+                    raise HTTPException(status_code=503, detail=error_msg)
+            
             # Create job for streaming using job manager
             job_id = job_manager.create_job(request_id=request_id, initial_data={
                 'request_id': request_id,
-                'message': 'GGUF model loading and generation in progress...' if is_gguf_mode else 'Job queued...'
+                'message': 'GGUF model loading and generation in progress...' if is_gguf_mode else 'Job queued...',
+                'queue_position': queue_manager.get_queue_status()['queue_size'] + 1 if stream else 0
             })
             job_manager.update_job(job_id, JOB_STATUS["QUEUED"], progress=0)
 
-            # Start background task
-            threading.Thread(target=process_patient_summary_background, args=(data, job_id), daemon=True).start()
+            # Wait for processing slot and start background task (for streaming only)
+            if stream:
+                def wait_and_process():
+                    """Wait for slot and then process."""
+                    if queue_manager.wait_for_slot(request_id, timeout=300):
+                        # Start background task
+                        threading.Thread(target=process_patient_summary_background, args=(data, job_id), daemon=True).start()
+                    else:
+                        # Timeout waiting for slot
+                        job_manager.update_job(job_id, JOB_STATUS["ERROR"], error="Request timed out waiting for processing slot")
+                        queue_manager.release_slot(request_id)
+                
+                # Start waiting thread
+                threading.Thread(target=wait_and_process, daemon=True).start()
+            else:
+                # Non-streaming: process immediately (no queue management needed)
+                pass
 
             # ALWAYS use extended generator to prevent timeout issues
             # Use SSE generator service directly

services/ai-service/src/ai_med_extract/services/__init__.py

@@ -0,0 +1,31 @@
+"""
+Services module for patient summary generation.
+"""
+
+from .job_manager import get_job_manager, JobManager
+from .error_handler import (
+    log_error_safely, handle_error_gracefully, update_job_with_error,
+    ErrorCategory, PatientSummaryError, categorize_error, create_error_response
+)
+from .sse_generator import sse_generator, sse_generator_extended, SSEGenerator
+from .request_queue import get_queue_manager, RequestQueueManager, RequestPriority, QueuedRequest
+
+__all__ = [
+    'get_job_manager',
+    'JobManager',
+    'log_error_safely',
+    'handle_error_gracefully',
+    'update_job_with_error',
+    'ErrorCategory',
+    'PatientSummaryError',
+    'categorize_error',
+    'create_error_response',
+    'sse_generator',
+    'sse_generator_extended',
+    'SSEGenerator',
+    'get_queue_manager',
+    'RequestQueueManager',
+    'RequestPriority',
+    'QueuedRequest',
+]
+

services/ai-service/src/ai_med_extract/services/request_queue.py

@@ -0,0 +1,303 @@
+"""
+Request Queue Manager for handling concurrent requests on Hugging Face Spaces.
+
+Provides request queuing, throttling, and resource management for limited-resource environments.
+"""
+
+import asyncio
+import time
+import threading
+from typing import Dict, Optional, Callable, Any
+from collections import deque
+from dataclasses import dataclass
+from enum import Enum
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class RequestPriority(Enum):
+    """Request priority levels."""
+    HIGH = 1
+    NORMAL = 2
+    LOW = 3
+
+
+@dataclass
+class QueuedRequest:
+    """Represents a queued request."""
+    request_id: str
+    job_id: str
+    priority: RequestPriority
+    timestamp: float
+    callback: Optional[Callable] = None
+
+
+class RequestQueueManager:
+    """
+    Manages concurrent requests with queuing and resource limits.
+    
+    Designed for Hugging Face Spaces with limited resources (T4 medium GPU).
+    """
+    
+    def __init__(
+        self,
+        max_concurrent: int = 2,  # T4 medium can handle 2 concurrent requests
+        max_queue_size: int = 10,
+        queue_timeout: int = 300  # 5 minutes max wait in queue
+    ):
+        """
+        Initialize request queue manager.
+        
+        Args:
+            max_concurrent: Maximum concurrent requests (default: 2 for T4 medium)
+            max_queue_size: Maximum queue size
+            queue_timeout: Maximum time to wait in queue (seconds)
+        """
+        self.max_concurrent = max_concurrent
+        self.max_queue_size = max_queue_size
+        self.queue_timeout = queue_timeout
+        
+        self._queue: deque = deque()
+        self._active_requests: Dict[str, float] = {}  # request_id -> start_time
+        self._lock = threading.RLock()
+        self._condition = threading.Condition(self._lock)
+        
+        # Statistics
+        self._total_processed = 0
+        self._total_rejected = 0
+        self._total_timeout = 0
+    
+    def can_accept_request(self) -> bool:
+        """
+        Check if a new request can be accepted.
+        
+        Returns:
+            True if request can be accepted
+        """
+        with self._lock:
+            active_count = len(self._active_requests)
+            queue_size = len(self._queue)
+            
+            # Check if we can accept immediately
+            if active_count < self.max_concurrent:
+                return True
+            
+            # Check if queue has space
+            if queue_size < self.max_queue_size:
+                return True
+            
+            return False
+    
+    def enqueue_request(
+        self,
+        request_id: str,
+        job_id: str,
+        priority: RequestPriority = RequestPriority.NORMAL
+    ) -> tuple[bool, Optional[str]]:
+        """
+        Enqueue a request for processing.
+        
+        Args:
+            request_id: Request identifier
+            job_id: Job identifier
+            priority: Request priority
+            
+        Returns:
+            Tuple of (accepted, error_message)
+        """
+        with self._lock:
+            # Check if we can process immediately
+            if len(self._active_requests) < self.max_concurrent:
+                self._active_requests[request_id] = time.time()
+                self._total_processed += 1
+                logger.info(f"Request {request_id} accepted immediately (active: {len(self._active_requests)})")
+                return True, None
+            
+            # Check queue capacity
+            if len(self._queue) >= self.max_queue_size:
+                self._total_rejected += 1
+                error_msg = f"Queue full ({len(self._queue)}/{self.max_queue_size}). Please try again later."
+                logger.warning(f"Request {request_id} rejected: {error_msg}")
+                return False, error_msg
+            
+            # Add to queue
+            queued_request = QueuedRequest(
+                request_id=request_id,
+                job_id=job_id,
+                priority=priority,
+                timestamp=time.time()
+            )
+            
+            # Insert based on priority
+            inserted = False
+            for i, req in enumerate(self._queue):
+                if priority.value < req.priority.value:
+                    self._queue.insert(i, queued_request)
+                    inserted = True
+                    break
+            
+            if not inserted:
+                self._queue.append(queued_request)
+            
+            logger.info(f"Request {request_id} queued (position: {len(self._queue)}, active: {len(self._active_requests)})")
+            
+            # Notify waiting threads
+            self._condition.notify_all()
+            
+            return True, None
+    
+    def wait_for_slot(self, request_id: str, timeout: Optional[int] = None) -> bool:
+        """
+        Wait for a processing slot to become available.
+        
+        Args:
+            request_id: Request identifier
+            timeout: Timeout in seconds (defaults to queue_timeout)
+            
+        Returns:
+            True if slot acquired, False if timeout
+        """
+        timeout = timeout or self.queue_timeout
+        start_time = time.time()
+        
+        with self._condition:
+            while True:
+                # Check if we can process now
+                if len(self._active_requests) < self.max_concurrent:
+                    # Remove from queue if present
+                    self._queue = deque([r for r in self._queue if r.request_id != request_id])
+                    self._active_requests[request_id] = time.time()
+                    logger.info(f"Request {request_id} acquired slot (active: {len(self._active_requests)})")
+                    return True
+                
+                # Check timeout
+                elapsed = time.time() - start_time
+                if elapsed >= timeout:
+                    # Remove from queue
+                    self._queue = deque([r for r in self._queue if r.request_id != request_id])
+                    self._total_timeout += 1
+                    logger.warning(f"Request {request_id} timed out waiting for slot ({elapsed:.1f}s)")
+                    return False
+                
+                # Wait for notification
+                remaining_timeout = timeout - elapsed
+                self._condition.wait(timeout=min(remaining_timeout, 5.0))
+    
+    def release_slot(self, request_id: str) -> None:
+        """
+        Release a processing slot.
+        
+        Args:
+            request_id: Request identifier
+        """
+        with self._lock:
+            if request_id in self._active_requests:
+                processing_time = time.time() - self._active_requests[request_id]
+                del self._active_requests[request_id]
+                logger.info(f"Request {request_id} released slot (processing time: {processing_time:.1f}s, active: {len(self._active_requests)})")
+                
+                # Notify waiting threads
+                self._condition.notify_all()
+    
+    def get_queue_status(self) -> Dict[str, Any]:
+        """
+        Get current queue status.
+        
+        Returns:
+            Dictionary with queue statistics
+        """
+        with self._lock:
+            return {
+                "active_requests": len(self._active_requests),
+                "queue_size": len(self._queue),
+                "max_concurrent": self.max_concurrent,
+                "max_queue_size": self.max_queue_size,
+                "total_processed": self._total_processed,
+                "total_rejected": self._total_rejected,
+                "total_timeout": self._total_timeout,
+                "queue_positions": [
+                    {
+                        "request_id": req.request_id,
+                        "job_id": req.job_id,
+                        "priority": req.priority.name,
+                        "wait_time": time.time() - req.timestamp
+                    }
+                    for req in self._queue
+                ]
+            }
+    
+    def cleanup_old_requests(self, max_age: int = 3600) -> int:
+        """
+        Clean up old requests from tracking.
+        
+        Args:
+            max_age: Maximum age in seconds
+            
+        Returns:
+            Number of requests cleaned up
+        """
+        with self._lock:
+            current_time = time.time()
+            cleaned = 0
+            
+            # Clean active requests
+            to_remove = [
+                req_id for req_id, start_time in self._active_requests.items()
+                if current_time - start_time > max_age
+            ]
+            
+            for req_id in to_remove:
+                del self._active_requests[req_id]
+                cleaned += 1
+            
+            # Clean queue
+            queue_size_before = len(self._queue)
+            self._queue = deque([
+                req for req in self._queue
+                if current_time - req.timestamp < max_age
+            ])
+            cleaned += queue_size_before - len(self._queue)
+            
+            if cleaned > 0:
+                logger.info(f"Cleaned up {cleaned} old requests")
+            
+            return cleaned
+
+
+# Global singleton instance
+_queue_manager: Optional[RequestQueueManager] = None
+
+
+def get_queue_manager() -> RequestQueueManager:
+    """
+    Get the global queue manager instance (singleton pattern).
+    
+    Returns:
+        RequestQueueManager instance
+    """
+    global _queue_manager
+    if _queue_manager is None:
+        # Detect HF Spaces and adjust limits
+        import os
+        is_hf_spaces = os.environ.get('HF_SPACES', 'false').lower() == 'true'
+        
+        if is_hf_spaces:
+            # T4 medium: 2 concurrent requests, queue of 5
+            _queue_manager = RequestQueueManager(
+                max_concurrent=2,
+                max_queue_size=5,
+                queue_timeout=300
+            )
+            logger.info("Initialized RequestQueueManager for Hugging Face Spaces (T4 medium)")
+        else:
+            # Local/dev: more generous limits
+            _queue_manager = RequestQueueManager(
+                max_concurrent=4,
+                max_queue_size=20,
+                queue_timeout=600
+            )
+            logger.info("Initialized RequestQueueManager for local/development")
+    
+    return _queue_manager
+