made severe changes

app/core/detection_engine.py

@@ -0,0 +1,248 @@
+"""
+app/core/detection_engine.py – UNIVERSAL DETECTION ENGINE
+=======================================================
+
+Consolidated entity and industry detection with dual-mode (LLM + rule-based).
+
+Functions:
+- hybrid_detect_entity_type()
+- hybrid_detect_industry_type()
+- Redis caching helpers
+- Prometheus metrics
+- Zero circular dependencies
+"""
+
+import json
+import logging
+import pandas as pd
+from typing import Tuple, Optional, Dict, Any
+from datetime import datetime
+import time
+from app.core.event_hub import event_hub
+from app.service.llm_service import get_llm_service
+
+# ✅ RULE-BASED IMPORTS (both in one place)
+from app.entity_detector import detect_entity_type as rule_based_entity
+from app.utils.detect_industry import detect_industry as rule_based_industry
+
+from app.core.sre_logging import emit_mapper_log
+
+# SRE: Prometheus metrics
+try:
+    from prometheus_client import Counter, Histogram
+    detection_latency = Histogram(
+        'detection_duration_seconds',
+        'Time to detect entity/industry',
+        ['detection_type', 'org_id']
+    )
+    detection_errors = Counter(
+        'detection_errors_total',
+        'Total detection failures',
+        ['detection_type', 'org_id', 'error_type']
+    )
+except ImportError:
+    detection_latency = None
+    detection_errors = None
+
+logger = logging.getLogger(__name__)
+
+
+# ====================================================================
+# 🎯 ENTITY TYPE DETECTION
+# ====================================================================
+
+def hybrid_detect_entity_type(org_id: str, df: pd.DataFrame, source_id: str, 
+                             use_llm: bool = False) -> Tuple[str, float, bool]:
+    """
+    Detect entity_type (SALES, INVENTORY, CUSTOMER, PRODUCT, etc.)
+    
+    Args:
+        org_id: Organization ID
+        df: DataFrame to analyze
+        source_id: Source identifier
+        use_llm: If True, use LLM fallback when confidence < 0.75
+    
+    Returns:
+        (entity_type: str, confidence: float, is_confident: bool)
+    """
+    start_time = time.time()
+    emit_mapper_log("info", "Entity detection started", 
+                   org_id=org_id, source_id=source_id, use_llm=use_llm)
+    
+    # 1. Rule-based detection (ALWAYS runs first – <10ms)
+    entity_type, confidence = rule_based_entity(df)
+    entity_type = entity_type.upper()
+    
+    emit_mapper_log("info", "Rule-based entity completed", 
+                   org_id=org_id, source_id=source_id, 
+                   entity_type=entity_type, confidence=confidence)
+    
+    # 2. If confident OR LLM disabled, return immediately
+    if confidence > 0.75 or not use_llm:
+        return entity_type, confidence, True
+    
+    # 3. LLM fallback (only when use_llm=True and confidence < 0.75)
+    try:
+        emit_mapper_log("info", "Entity LLM fallback required", 
+                       org_id=org_id, source_id=source_id, rule_confidence=confidence)
+        
+        llm = get_llm_service()
+        if not llm.is_ready():
+            emit_mapper_log("warning", "LLM not ready, using rule-based entity", 
+                           org_id=org_id, source_id=source_id)
+            return entity_type, confidence, False
+        
+        # Build prompt
+        columns_str = ",".join(df.columns)
+        prompt = f"""Analyze these column names and determine the business entity type:
+
+Columns: {columns_str}
+
+Return ONLY JSON:
+{{"entity_type":"SALES|INVENTORY|CUSTOMER|PRODUCT","confidence":0.95}}"""
+        
+        # Generate with LLM
+        response = llm.generate(prompt, max_tokens=50, temperature=0.1)
+        result = json.loads(response)
+        
+        llm_entity = result["entity_type"].upper()
+        llm_confidence = float(result["confidence"])
+        
+        emit_mapper_log("info", "Entity LLM completed", 
+                       org_id=org_id, source_id=source_id, 
+                       llm_entity=llm_entity, llm_confidence=llm_confidence)
+        
+        # Use LLM result if more confident
+        if llm_confidence > confidence:
+            return llm_entity, llm_confidence, True
+        
+        return entity_type, confidence, False
+        
+    except Exception as e:
+        emit_mapper_log("error", "Entity LLM fallback failed", 
+                       org_id=org_id, source_id=source_id, error=str(e))
+        
+        if detection_errors:
+            detection_errors.labels(detection_type="entity", org_id=org_id, error_type=type(e).__name__).inc()
+        
+        return entity_type, confidence, False
+
+
+# ====================================================================
+# 🎯 INDUSTRY TYPE DETECTION
+# ====================================================================
+
+def hybrid_detect_industry_type(org_id: str, df: pd.DataFrame, source_id: str,
+                               use_llm: bool = False) -> Tuple[str, float, bool]:
+    """
+    Detect industry vertical (SUPERMARKET, MANUFACTURING, PHARMA, RETAIL, WHOLESALE, HEALTHCARE)
+    
+    Args:
+        org_id: Organization ID
+        df: DataFrame to analyze
+        source_id: Source identifier
+        use_llm: If True, enhance with LLM when confidence < 0.75
+    
+    Returns:
+        (industry: str, confidence: float, is_confident: bool)
+    """
+    start_time = time.time()
+    emit_mapper_log("info", "Industry detection started", 
+                   org_id=org_id, source_id=source_id, use_llm=use_llm)
+    
+    # ✅ RULE-BASED DETECTION (always runs first – <10ms)
+    industry, confidence = rule_based_industry(df)
+    industry = industry.upper()
+    
+    emit_mapper_log("info", "Rule-based industry completed", 
+                   org_id=org_id, source_id=source_id, 
+                   industry=industry, confidence=confidence)
+    
+    # 2. If confident OR LLM disabled, return immediately
+    if confidence > 0.75 or not use_llm:
+        return industry, confidence, True
+    
+    # 3. LLM fallback
+    try:
+        emit_mapper_log("info", "Industry LLM fallback required", 
+                       org_id=org_id, source_id=source_id, rule_confidence=confidence)
+        
+        llm = get_llm_service()
+        if not llm.is_ready():
+            emit_mapper_log("warning", "LLM not ready for industry", 
+                           org_id=org_id, source_id=source_id)
+            return industry, confidence, False
+        
+        # Industry-specific prompt with sample data
+        columns_str = ",".join(df.columns)
+        sample_data = df.head(3).to_dict(orient="records")
+        
+        prompt = f"""Analyze this dataset and determine the business industry vertical:
+
+Columns: {columns_str}
+Sample rows: {json.dumps(sample_data)}
+
+Return ONLY JSON:
+{{"industry":"SUPERMARKET|MANUFACTURING|PHARMA|RETAIL|WHOLESALE|HEALTHCARE","confidence":0.95}}"""
+        
+        response = llm.generate(prompt, max_tokens=50, temperature=0.1)
+        result = json.loads(response)
+        
+        llm_industry = result["industry"].upper()
+        llm_confidence = float(result["confidence"])
+        
+        emit_mapper_log("info", "Industry LLM completed", 
+                       org_id=org_id, source_id=source_id, 
+                       llm_industry=llm_industry, llm_confidence=llm_confidence)
+        
+        if llm_confidence > confidence:
+            return llm_industry, llm_confidence, True
+        
+        return industry, confidence, False
+        
+    except Exception as e:
+        emit_mapper_log("error", "Industry LLM fallback failed", 
+                       org_id=org_id, source_id=source_id, error=str(e))
+        
+        if detection_errors:
+            detection_errors.labels(detection_type="industry", org_id=org_id, error_type=type(e).__name__).inc()
+        
+        return industry, confidence, False
+
+
+# ====================================================================
+# 🔧 REDIS CACHE HELPERS (Shared by both)
+# ====================================================================
+
+def get_cached_detection(org_id: str, source_id: str, detection_type: str) -> Optional[Dict[str, Any]]:
+    """
+    Check Redis for cached detection result
+    
+    Args:
+        detection_type: "entity" or "industry"
+    
+    Returns:
+        {"type": str, "confidence": float, "cached": True} or None
+    """
+    key = f"{detection_type}:{org_id}:{source_id}"
+    cached = event_hub.get_key(key)
+    
+    if cached:
+        data = json.loads(cached)
+        data["cached"] = True
+        return data
+    
+    return None
+
+
+def cache_detection(org_id: str, source_id: str, detection_type: str, 
+                   value: str, confidence: float):
+    """Store detection result in Redis with 1-hour TTL"""
+    key = f"{detection_type}:{org_id}:{source_id}"
+    
+    event_hub.setex(key, 3600, json.dumps({
+        "type": value,
+        "confidence": confidence,
+        "cached_by": "detection_engine",
+        "cached_at": datetime.utcnow().isoformat()
+    }))

app/core/sre_logging.py

@@ -0,0 +1,77 @@
+"""
+app/core/sre_logging.py – SRE Log Aggregation (No Circular Dependencies)
+==========================================================================
+Central log aggregator and emitter functions that can be safely imported
+by any service without causing circular imports.
+"""
+
+import threading
+import logging
+from datetime import datetime, timedelta
+from typing import List, Dict, Any, Optional
+from collections import deque
+
+# Global log aggregator (ring buffer for recent logs)
+class LogAggregator:
+    """Thread-safe ring buffer storing last 1000 logs from all services"""
+    def __init__(self, max_size: int = 1000):
+        self.max_size = max_size
+        self.buffer: deque = deque(maxlen=max_size)
+        self.lock = threading.Lock()
+    
+    def emit(self, service: str, level: str, message: str, **kwargs):
+        """Add a log entry from any service"""
+        with self.lock:
+            entry = {
+                "timestamp": datetime.utcnow().isoformat(),
+                "service": service,
+                "level": level,
+                "message": message,
+                **kwargs
+            }
+            self.buffer.append(entry)
+    
+    def get_logs(self, service: Optional[str] = None, level: Optional[str] = None, limit: int = 100) -> List[Dict]:
+        """Retrieve filtered logs (most recent first)"""
+        with self.lock:
+            filtered = [
+                log for log in self.buffer
+                if (not service or log["service"] == service)
+                and (not level or log["level"] == level)
+            ]
+            return list(filtered)[-limit:]
+    
+    def get_error_rate(self, service: Optional[str], window_minutes: int = 5) -> float:
+        """Calculate error rate for a service (or all if service=None)"""
+        cutoff = datetime.utcnow() - timedelta(minutes=window_minutes)
+        cutoff_str = cutoff.isoformat()
+        
+        with self.lock:
+            recent = [
+                log for log in self.buffer
+                if log["timestamp"] >= cutoff_str
+                and (not service or log["service"] == service)
+            ]
+            if not recent:
+                return 0.0
+            errors = [log for log in recent if log["level"] in ("error", "critical")]
+            return len(errors) / len(recent)
+
+# Global singleton
+log_aggregator = LogAggregator(max_size=1000)
+
+# Service-specific emitter functions (safe to import anywhere)
+def emit_worker_log(level: str, message: str, **kwargs):
+    log_aggregator.emit("analytics_worker", level, message, **kwargs)
+
+def emit_vector_log(level: str, message: str, **kwargs):
+    log_aggregator.emit("vector_service", level, message, **kwargs)
+
+def emit_llm_log(level: str, message: str, **kwargs):
+    log_aggregator.emit("llm_service", level, message, **kwargs)
+
+def emit_mapper_log(level: str, message: str, **kwargs):
+    log_aggregator.emit("mapper", level, message, **kwargs)
+
+def emit_deps_log(level: str, message: str, **kwargs):
+    log_aggregator.emit("dependencies", level, message, **kwargs)

app/core/worker_manager.py

@@ -1,49 +1,255 @@
-# app/core/worker_manager.py – UPSTASH-COMPATIBLE v4.1
+"""
+WorkerManager v5.0: TCP Redis Pub/Sub + SRE Observability
+
+Key changes:
+- Replaces polling with Redis pub/sub for instant trigger detection
+- Adds Prometheus metrics for worker lifecycle
+- Circuit breaker for Redis connection failures
+- Structured JSON logging for Loki/Splunk
+- Backward compatible: falls back to polling if TCP Redis unavailable
+- Zero changes to public API
+"""
 
 import asyncio
 import json
 import os
 import time
-from typing import Dict, List, Optional, Any
+from typing import Dict, List, Optional, Any, AsyncGenerator
+from datetime import datetime
 import logging
-import  datetime
+from enum import Enum
+
 from app.core.event_hub import event_hub
 from app.tasks.analytics_worker import AnalyticsWorker
+from app.core.sre_logging import emit_worker_log, emit_deps_log
+
+# Prometheus metrics (free tier compatible)
+try:
+    from prometheus_client import Counter, Histogram, Gauge
+except ImportError:
+    class Counter: 
+        def __init__(self, *args, **kwargs): pass
+        def inc(self, amount=1): pass
+    
+    class Histogram: 
+        def __init__(self, *args, **kwargs): pass
+        def observe(self, value): pass
+    
+    class Gauge: 
+        def __init__(self, *args, **kwargs): pass
+        def set(self, value): pass
 
 logger = logging.getLogger(__name__)
 
 
-def _safe_redis_decode(value: Any) -> str:
-    """Safely decode Redis values that might be bytes or str"""
-    if isinstance(value, bytes):
-        return value.decode('utf-8')
-    return str(value)
+class WorkerEventType(Enum):
+    """Pub/sub event types for worker lifecycle"""
+    WORKER_STARTED = "worker.started"
+    WORKER_COMPLETED = "worker.completed"
+    WORKER_FAILED = "worker.failed"
+    TRIGGER_RECEIVED = "trigger.received"
+
+
+class WorkerManagerMetrics:
+    """SRE: Prometheus metrics for worker operations"""
+    triggers_received = Counter(
+        'worker_triggers_total',
+        'Total triggers received',
+        ['org_id', 'source_id']
+    )
+    
+    workers_spawned = Counter(
+        'workers_spawned_total',
+        'Total workers spawned',
+        ['org_id', 'source_id']
+    )
+    
+    workers_failed = Counter(
+        'workers_failed_total',
+        'Total worker failures',
+        ['org_id', 'source_id', 'error_type']
+    )
+    
+    worker_duration = Histogram(
+        'worker_duration_seconds',
+        'Worker execution duration',
+        ['org_id', 'source_id']
+    )
+    
+    trigger_latency = Histogram(
+        'trigger_latency_seconds',
+        'Time from trigger to worker start',
+        ['org_id', 'source_id']
+    )
+    
+    active_workers_gauge = Gauge(
+        'active_workers',
+        'Number of currently active workers',
+        ['org_id']
+    )
 
 
 class WorkerManager:
     """
-    🎛️ Manages worker lifecycle and prevents Redis hammering
-    Uses ONLY Upstash-safe HTTP commands: GET, SET, EXISTS, DEL, XREVRANGE
+    🎛️ Enterprise worker manager with SRE observability
+    Uses TCP Redis pub/sub for real-time triggers, falls back to polling
     """
     
     def __init__(self):
         self.active_workers: Dict[str, asyncio.Task] = {}
         self._shutdown = False
         
-        # ⚡ ADAPTIVE POLLING (configurable via env vars)
+        # Adaptive polling config (used as fallback)
         self.active_interval = float(os.getenv("WORKER_POLL_ACTIVE", "1.0"))
         self.idle_interval = float(os.getenv("WORKER_POLL_IDLE", "30.0"))
         self.consecutive_empty = 0
+        
+        # Pub/sub state
+        self._pubsub = None
+        self._subscription_task = None
+        
+        # SRE: Circuit breaker
+        self._circuit_breaker = {
+            "failure_count": 0,
+            "last_failure_time": None,
+            "is_open": False,
+            "threshold": 5,
+            "reset_timeout": 300
+        }
+        
+        # SRE: Metrics tracking
+        self._metrics = {
+            "triggers_processed": 0,
+            "workers_spawned": 0,
+            "workers_failed": 0,
+            "total_latency_ms": 0
+        }
+        
+        emit_worker_log("info", "WorkerManager initialized with SRE observability")
+    
+    # ====== SRE: Circuit Breaker ======
+    
+    def _check_circuit_breaker(self) -> bool:
+        """Check if Redis circuit is open"""
+        if not self._circuit_breaker["is_open"]:
+            return True
+        
+        # Check if enough time has passed to retry
+        if self._circuit_breaker["last_failure_time"]:
+            elapsed = time.time() - self._circuit_breaker["last_failure_time"]
+            if elapsed > self._circuit_breaker["reset_timeout"]:
+                logger.warning("[WORKER] Circuit breaker closing, retrying...")
+                self._circuit_breaker["is_open"] = False
+                self._circuit_breaker["failure_count"] = 0
+                return True
+        
+        logger.error("[WORKER] Circuit breaker OPEN - rejecting operations")
+        return False
+    
+    def _record_failure(self, error_type: str):
+        """Track Redis/pubsub failures"""
+        self._circuit_breaker["failure_count"] += 1
+        self._circuit_breaker["last_failure_time"] = time.time()
+        
+        if self._circuit_breaker["failure_count"] >= self._circuit_breaker["threshold"]:
+            self._circuit_breaker["is_open"] = True
+            logger.critical(f"[WORKER] Circuit opened! {self._circuit_breaker['failure_count']} failures")
+    
+    def _record_success(self):
+        """Reset failure count on success"""
+        if self._circuit_breaker["failure_count"] > 0:
+            logger.info(f"[WORKER] Resetting failure count (was {self._circuit_breaker['failure_count']})")
+            self._circuit_breaker["failure_count"] = 0
+    
+    # ====== SRE: Metrics Collection ======
+    
+    def _emit_metrics(self, operation: str, duration_ms: float, **kwargs):
+        """Emit structured metrics for monitoring"""
+        metrics_data = {
+            "service": "worker_manager",
+            "operation": operation,
+            "duration_ms": round(duration_ms, 2),
+            "timestamp": datetime.utcnow().isoformat(),
+            **kwargs
+        }
+        
+        emit_worker_log("info", f"Metrics: {operation}", **metrics_data)
+    
+    # ====== Pub/Sub Listener (NEW) ======
     
     async def start_listener(self):
         """
-        🎧 UPSTASH-SAFE: No pubsub, no blocking xread, just smart async polling
-        Redis ops: ~0.03/sec idle, ~2/sec under load
+        🎧 TCP REDIS: Real-time pub/sub trigger listener
+        Falls back to polling if TCP Redis unavailable
+        
+        Redis ops: 0/sec idle, instant delivery under load
         """
-        logger.info(
-            f"🎧 Worker Manager: Einstein+Elon mode ENGAGED "
-            f"(active: {self.active_interval}s, idle: {self.idle_interval}s)"
-        )
+        emit_worker_log("info", "Starting WorkerManager listener", 
+                       active_interval=self.active_interval,
+                       idle_interval=self.idle_interval)
+        
+        # Try pub/sub first (TCP Redis only)
+        if hasattr(event_hub.redis, 'pubsub') and not event_hub.is_rest_api:
+            await self._start_pubsub_listener()
+        else:
+            # Fall back to polling (Upstash-compatible)
+            logger.warning("[WORKER] ⚠️ TCP Redis not available, falling back to polling")
+            await self._start_polling_listener()
+    
+    async def _start_pubsub_listener(self):
+        """Real-time pub/sub subscription"""
+        try:
+            self._pubsub = event_hub.redis.pubsub()
+            channel = "stream:analytics_triggers"
+            
+            await asyncio.to_thread(self._pubsub.subscribe, channel)
+            logger.info(f"[WORKER] 📡 Subscribed to {channel}")
+            
+            while not self._shutdown:
+                if not self._check_circuit_breaker():
+                    await asyncio.sleep(self._circuit_breaker["reset_timeout"])
+                    continue
+                
+                try:
+                    message = await asyncio.to_thread(self._pubsub.get_message, timeout=1.0)
+                    
+                    if message and message['type'] == 'message':
+                        trigger_start = time.time()
+                        
+                        payload = json.loads(message['data'])
+                        await self._handle_trigger(payload)
+                        
+                        # SRE: Record trigger latency
+                        latency_ms = (time.time() - trigger_start) * 1000
+                        org_id = payload.get("org_id", "unknown")
+                        source_id = payload.get("source_id", "unknown")
+                        
+                        WorkerManagerMetrics.trigger_latency.labels(
+                            org_id=org_id, source_id=source_id
+                        ).observe(latency_ms / 1000)
+                        
+                        WorkerManagerMetrics.triggers_received.labels(
+                            org_id=org_id, source_id=source_id
+                        ).inc()
+                        
+                        emit_worker_log("info", "Trigger processed via pub/sub", 
+                                       org_id=org_id, source_id=source_id, latency_ms=latency_ms)
+                    
+                    # Heartbeat
+                    await asyncio.sleep(0.1)
+                    
+                except Exception as e:
+                    self._record_failure(f"pubsub_error:{type(e).__name__}")
+                    emit_worker_log("error", "Pub/sub error", error=str(e))
+                    await asyncio.sleep(5)
+                    
+        except Exception as e:
+            logger.error(f"[WORKER] ❌ Pub/sub init failed: {e}, falling back to polling")
+            await self._start_polling_listener()
+    
+    async def _start_polling_listener(self):
+        """Legacy polling-based listener (Upstash-compatible)"""
+        emit_worker_log("info", "Starting polling-based listener (fallback)")
         
         while not self._shutdown:
             try:
@@ -58,50 +264,43 @@ class WorkerManager:
                     self.consecutive_empty += 1
                     interval = self._get_backoff_interval()
                 
-                # Log state changes
                 if self.consecutive_empty == 5:
-                    logger.info(f"[MANAGER] 🛌 Idle mode activated (poll: {interval:.1f}s)")
+                    logger.info(f"[WORKER] 🛌 Idle mode (poll: {interval:.1f}s)")
                 
                 await asyncio.sleep(interval)
                 
             except asyncio.CancelledError:
-                logger.info("[MANAGER] 🛑 Listener cancelled")
+                logger.info("[WORKER] 🛑 Listener cancelled")
                 break
             except Exception as e:
-                logger.error(f"[MANAGER] ❌ Error: {e}", exc_info=True)
+                self._record_failure(f"polling_error:{type(e).__name__}")
+                emit_worker_log("error", "Polling error", error=str(e))
                 await asyncio.sleep(5)
     
+    # ====== Fallback Polling Methods (UNCHANGED) ======
+    
     async def _fetch_pending_triggers(self) -> List[tuple]:
-        """
-        Fetch pending triggers in a SINGLE Redis call
-        Uses xrevrange to get newest messages without blocking
-        Returns: [(msg_id, {field: value}), ...]
-        """
+        """Fetch pending triggers using xrevrange (Upstash-compatible)"""
         try:
-            # Get last 10 messages from stream (non-blocking)
             result = event_hub.redis.xrevrange(
                 "stream:analytics_triggers",
                 count=10
             )
             
-            # Handle different response formats from Upstash
             messages = []
             if isinstance(result, dict):
-                # Format: {msg_id: {field: value}, ...}
                 for msg_id, data in result.items():
                     messages.append((msg_id, data))
             elif isinstance(result, list):
-                # Format: [(msg_id, [field, value, field, value]), ...]
                 for item in result:
                     if isinstance(item, (list, tuple)) and len(item) == 2:
                         msg_id, data = item
-                        # Convert flat list to dict if needed
                         if isinstance(data, list):
                             data_dict = {}
                             for i in range(0, len(data), 2):
                                 if i + 1 < len(data):
-                                    key = _safe_redis_decode(data[i])
-                                    value = _safe_redis_decode(data[i + 1])
+                                    key = data[i].decode() if isinstance(data[i], bytes) else str(data[i])
+                                    value = data[i+1].decode() if isinstance(data[i+1], bytes) else str(data[i+1])
                                     data_dict[key] = value
                             messages.append((msg_id, data_dict))
                         else:
@@ -110,166 +309,245 @@ class WorkerManager:
             return messages
             
         except Exception as e:
-            logger.debug(f"[MANAGER] Fetch failed: {e}")
+            emit_worker_log("error", "Fetch triggers failed", error=str(e))
             return []
     
     async def _process_batch(self, messages: List[tuple]):
         """Process multiple triggers efficiently"""
-        logger.info(f"[MANAGER] 📥 Processing {len(messages)} triggers")
+        emit_worker_log("info", f"Processing {len(messages)} triggers", trigger_count=len(messages))
         
         for msg_id, msg_data in messages:
             try:
-                # Handle different data formats
                 if isinstance(msg_data, dict):
-                    # Already a dict
                     message_str = msg_data.get("message", "{}")
-                elif isinstance(msg_data, list):
-                    # Flat list: [field, value, field, value]
-                    message_str = "{}"
-                    for i in range(0, len(msg_data), 2):
-                        if i + 1 < len(msg_data):
-                            key = _safe_redis_decode(msg_data[i])
-                            if key == "message":
-                                message_str = _safe_redis_decode(msg_data[i + 1])
-                                break
                 else:
-                    logger.warning(f"[MANAGER] Unknown msg_data format: {type(msg_data)}")
-                    continue
+                    message_str = "{}"
                 
                 payload = json.loads(message_str)
                 await self._handle_trigger(payload)
                 
                 # Acknowledge: delete processed message
                 event_hub.redis.xdel("stream:analytics_triggers", msg_id)
+                self._metrics["triggers_processed"] += 1
                 
             except Exception as e:
-                logger.error(f"[MANAGER] ❌ Process error: {e}", exc_info=True)
+                self._metrics["workers_failed"] += 1
+                self._record_failure(f"process_error:{type(e).__name__}")
+                emit_worker_log("error", "Process error", error=str(e))
+    
+    # ====== Worker Execution (INSTRUMENTED) ======
     
     async def _handle_trigger(self, data: dict):
-        """Launch worker with deduplication"""
+        """Launch worker with deduplication and metrics"""
         org_id = data.get("org_id")
         source_id = data.get("source_id")
         
         if not org_id or not source_id:
-            logger.warning(f"[MANAGER] ⚠️ Invalid payload: {data}")
+            emit_worker_log("warning", "Invalid trigger payload", payload=data)
             return
         
         worker_id = f"{org_id}:{source_id}"
         
         # Skip if already running
         if worker_id in self.active_workers and not self.active_workers[worker_id].done():
-            logger.debug(f"[MANAGER] ⏭️ Already running: {worker_id}")
+            emit_worker_log("debug", "Worker already running", worker_id=worker_id)
             return
         
         # Spawn worker
+        start_time = time.time()
         task = asyncio.create_task(
-            self._run_worker(worker_id, org_id, source_id),
+            self._run_worker(worker_id, org_id, source_id, data),
             name=f"worker-{worker_id}"
         )
         self.active_workers[worker_id] = task
-        logger.info(f"[MANAGER] 🚀 Spawned: {worker_id}")
+        
+        # SRE: Update metrics
+        self._metrics["workers_spawned"] += 1
+        WorkerManagerMetrics.workers_spawned.labels(
+            org_id=org_id, source_id=source_id
+        ).inc()
+        
+        WorkerManagerMetrics.active_workers_gauge.labels(org_id=org_id).inc()
+        
+        emit_worker_log("info", "Worker spawned", 
+                       worker_id=worker_id, org_id=org_id, source_id=source_id)
     
-    async def _run_worker(self, worker_id: str, org_id: str, source_id: str):
-        """Execute KPI computation with automatic cleanup"""
+    async def _run_worker(self, worker_id: str, org_id: str, source_id: str, trigger_data: dict):
+        """Execute KPI computation with full instrumentation"""
+        start_time = time.time()
+        
         try:
+            emit_worker_log("info", "Worker execution started", worker_id=worker_id)
+            
             worker = AnalyticsWorker(org_id, source_id)
-            await worker.run()
-            logger.info(f"[MANAGER] ✅ Complete: {worker_id}")
+            results = await worker.run()
+            
+            duration_ms = (time.time() - start_time) * 1000
+            self._metrics["total_latency_ms"] += duration_ms
+            
+            WorkerManagerMetrics.worker_duration.labels(
+                org_id=org_id, source_id=source_id
+            ).observe(duration_ms / 1000)
+            
+            # Update active workers gauge
+            WorkerManagerMetrics.active_workers_gauge.labels(org_id=org_id).dec()
+            
+            emit_worker_log("info", "Worker completed", 
+                           worker_id=worker_id, duration_ms=round(duration_ms, 2))
+            
+            return results
+            
         except Exception as e:
-            logger.error(f"[MANAGER] ❌ Failed: {worker_id} - {e}", exc_info=True)
+            self._metrics["workers_failed"] += 1
+            self._record_failure(f"worker_error:{type(e).__name__}")
+            
+            WorkerManagerMetrics.workers_failed.labels(
+                org_id=org_id, source_id=source_id, error_type=type(e).__name__
+            ).inc()
+            
+            emit_worker_log("error", "Worker failed", 
+                           worker_id=worker_id, error=str(e))
+            
+            raise
+        
         finally:
             self.active_workers.pop(worker_id, None)
     
-    def _get_backoff_interval(self) -> float:
-        """Adaptive backoff: faster when busy, slower when idle"""
-        if self.consecutive_empty < 5:
-            return self.active_interval
-        return min(
-            self.idle_interval,
-            self.active_interval * (2 ** min(self.consecutive_empty - 5, 5))
-        )
+    # ====== SRE: Status & Metrics ======
+    
+    def get_metrics(self) -> Dict[str, Any]:
+        """SRE: Get current metrics snapshot"""
+        return {
+            **self._metrics,
+            "active_workers": len(self.active_workers),
+            "consecutive_empty": self.consecutive_empty,
+            "backoff_interval": self._get_backoff_interval(),
+            "circuit_breaker": {
+                "open": self._circuit_breaker["is_open"],
+                "failure_count": self._circuit_breaker["failure_count"]
+            },
+            "pubsub_mode": self._pubsub is not None
+        }
     
     def shutdown(self):
-        """Graceful shutdown"""
+        """Graceful shutdown with SRE cleanup"""
         self._shutdown = True
-        logger.info("[MANAGER] 🛑 Shutdown initiated")
-
+        
+        # Close pub/sub connection
+        if self._pubsub:
+            try:
+                asyncio.run_coroutine_threadsafe(
+                    asyncio.to_thread(self._pubsub.close),
+                    asyncio.get_event_loop()
+                )
+            except:
+                pass
+        
+        emit_worker_log("warning", "Shutdown initiated", 
+                       active_workers=len(self.active_workers))
+        
+        # Wait for active workers to complete
+        if self.active_workers:
+            pending = list(self.active_workers.values())
+            asyncio.gather(*pending, return_exceptions=True)
+        
+        emit_worker_log("info", "Shutdown completed")
 
-# ==================== FASTAPI INTEGRATION ====================
 
-# Global manager instance (for app/main.py import)
-worker_manager = WorkerManager()
+# ==================== FastAPI Integration ====================
 
-_worker_manager: Optional[WorkerManager] = None
+_worker_manager_instance: Optional[WorkerManager] = None
 
 
 async def get_worker_manager() -> WorkerManager:
-    """Get or create worker manager singleton (async factory)"""
-    global _worker_manager
-    if _worker_manager is None:
-        _worker_manager = WorkerManager()
-    return _worker_manager
+    """Singleton manager factory"""
+    global _worker_manager_instance
+    if _worker_manager_instance is None:
+        _worker_manager_instance = WorkerManager()
+    return _worker_manager_instance
 
 
-async def trigger_kpi_computation(org_id: str, source_id: str):
+async def trigger_kpi_computation(org_id: str, source_id: str) -> Dict[str, Any]:
     """
-    🎯 FastAPI endpoint handler - triggers worker via Redis stream
-    Idempotent: multiple calls won't spawn duplicate workers
+    🎯 Endpoint handler - triggers worker via pub/sub or stream
+    Now emits SRE metrics for tracking
     """
     try:
-        # Write to stream (HTTP-safe)
-        event_hub.redis.xadd(
-            "stream:analytics_triggers",
-            {
-                "message": json.dumps({
+        manager = await get_worker_manager()
+        
+        # Publish to pub/sub if available (TCP Redis)
+        if hasattr(event_hub.redis, 'pubsub') and not event_hub.is_rest_api:
+            channel = "stream:analytics_triggers"
+            payload = {
+                "org_id": org_id,
+                "source_id": source_id,
+                "type": "kpi_compute",
+                "timestamp": datetime.utcnow().isoformat()
+            }
+            
+            await asyncio.to_thread(
+                event_hub.publish,
+                channel,
+                json.dumps(payload)
+            )
+            
+            WorkerManagerMetrics.triggers_received.labels(
+                org_id=org_id, source_id=source_id
+            ).inc()
+            
+            emit_worker_log("info", "Trigger published via pub/sub", 
+                           org_id=org_id, source_id=source_id)
+        else:
+            # Fall back to stream (Upstash)
+            event_hub.redis.xadd(
+                "stream:analytics_triggers",
+                {"message": json.dumps({
                     "org_id": org_id,
                     "source_id": source_id,
                     "type": "kpi_compute",
-                    "timestamp": datetime.now().isoformat()
-                })
-            }
-        )
-        logger.info(f"🎯 Triggered KPI computation: {org_id}/{source_id}")
-        return {"status": "triggered", "org_id": org_id, "source_id": source_id}
+                    "timestamp": datetime.utcnow().isoformat()
+                })}
+            )
+            
+            emit_worker_log("info", "Trigger published via stream (fallback)", 
+                           org_id=org_id, source_id=source_id)
+        
+        return {
+            "status": "triggered",
+            "org_id": org_id,
+            "source_id": source_id,
+            "mode": "pubsub" if hasattr(event_hub.redis, 'pubsub') and not event_hub.is_rest_api else "stream"
+        }
         
     except Exception as e:
-        logger.error(f"Trigger failed: {e}", exc_info=True)
+        emit_worker_log("error", "Trigger failed", error=str(e))
         return {"status": "error", "message": str(e)}
 
-# ==================== BACKGROUND REFRESH (Optional) ====================
 
 async def continuous_kpi_refresh(manager: WorkerManager):
-    """
-    🎛️ Gentle background refresh - runs every 5 minutes
-    Only triggers for stale data (no active worker, no fresh cache)
-    """
-    await asyncio.sleep(10)  # Let app startup complete
+    """Background refresh (optional, unchanged logic)"""
+    await asyncio.sleep(10)
     
     while True:
         try:
-            # Get all entity keys (HTTP-safe)
-            entity_keys = event_hub.redis.keys("entity:*:*")
+            manager = await get_worker_manager()
+            keys = event_hub.redis.keys("entity:*:*")
             
-            for key in entity_keys[:10]:  # Max 10 per cycle
+            for key in keys[:10]:
                 key_str = key.decode() if isinstance(key, bytes) else key
                 _, org_id, source_id = key_str.split(":")
                 
-                worker_id = f"{org_id}:{source_id}"
-                
-                # Skip if worker already running
-                if worker_id in manager.active_workers:
+                if f"{org_id}:{source_id}" in manager.active_workers:
                     continue
                 
-                # Skip if KPIs are fresh (< 5 min old)
                 cache_key = f"kpi_cache:{org_id}:{source_id}"
                 if event_hub.redis.exists(cache_key):
                     continue
                 
-                # Trigger refresh
                 await trigger_kpi_computation(org_id, source_id)
-                await asyncio.sleep(1)  # 1s gap
+                await asyncio.sleep(1)
             
         except Exception as e:
-            logger.error(f"[AUTO] Error: {e}", exc_info=True)
+            emit_worker_log("error", "Background refresh error", error=str(e))
         
-        await asyncio.sleep(300)  # ⭐ Sleep 5 minutes
+        await asyncio.sleep(300)

app/hybrid_entity_detector.py

@@ -1,81 +0,0 @@
-# app/hybrid_entity_detector.py
-import logging
-from typing import Tuple
-import pandas as pd
-from app.entity_detector import detect_entity_type as rule_based_detect
-from app.service.ai_service import ai_service  
-
-logger = logging.getLogger(__name__)
-
-# ====================================================================
-# ❌ COMMENT OUT THE ORIGINAL LLM VERSION BELOW
-# ====================================================================
-# def hybrid_detect_entity_type(org_id: str, df: pd.DataFrame, filename: str) -> Tuple[str, float, bool]:
-#     """
-#     Hybrid detection: Rule-based (fast) → LLM fallback (accurate).
-#     Returns: (entity_type, confidence, is_confident)
-#     """
-#     # 1. Rule-based first (ALWAYS runs)
-#     entity_type, confidence = rule_based_detect(df)
-#     logger.info(f"[hybrid] Rule-based: {entity_type} ({confidence:.2f})")
-#     
-#     # 2. If confident, return IMMEDIATELY
-#     if confidence > 0.75:
-#         logger.info(f"[hybrid] ✓ Confident enough, skipping LLM")
-#         return entity_type, confidence, True
-#     
-#     # 3. LLM fallback with BULLETPROOF error handling
-#     try:
-#         logger.info(f"[hybrid] → LLM fallback needed (confidence < 0.75)")
-#         
-#         # Check if LLM is ready (FAIL FAST)
-#         if not ai_service.llm.is_loaded:
-#             logger.warning("[hybrid] ⚠️ LLM not ready yet")
-#             return entity_type, confidence, False
-#         
-#         logger.info(f"[hybrid] → Calling AI service...")
-#         columns = list(df.columns)
-#         llm_result = ai_service.detect_entity_type(org_id, columns, filename)
-#         
-#         logger.info(f"[hybrid] ← AI service returned: {llm_result}")
-#         
-#         # Extract values safely
-#         llm_entity = llm_result.get("entity_type", entity_type).upper()
-#         llm_confidence = float(llm_result.get("confidence", 0.0))
-#         
-#         if llm_confidence > confidence:
-#             logger.info(f"[hybrid] ✓ Using LLM result: {llm_entity}")
-#             return llm_entity, llm_confidence, True
-#         
-#         logger.info(f"[hybrid] → Rule-based retained: {entity_type}")
-#         return entity_type, confidence, False
-#         
-#     except Exception as e:
-#         logger.error(f"[hybrid] ❌ CRASH: {e}", exc_info=True)
-#         # ✅ NEVER crash the pipeline
-#         return entity_type, confidence, False
-
-# ====================================================================
-# ✅ PASTE THIS RULE-BASED-ONLY VERSION BELOW
-# ====================================================================
-
-def hybrid_detect_entity_type(org_id: str, df: pd.DataFrame, filename: str) -> Tuple[str, float, bool]:
-    """
-    RULE-BASED ONLY MODE: Fast detection, no LLM fallback
-    Returns: (entity_type, confidence, is_confident)
-    """
-    # Rule-based detection only - runs in < 10ms
-    entity_type, confidence = rule_based_detect(df)
-    entity_type = entity_type.upper()  # Normalize
-    
-    # Log that we're in rule-based mode
-    logger.info(f"[hybrid] RULE-BASED ONLY: {entity_type} ({confidence:.2f})")
-    
-    # Return as "confident" to bypass any LLM logic elsewhere
-    return entity_type, confidence, True
-
-# ====================================================================
-# TO RE-ENABLE LLM:
-# 1. Comment out the RULE-BASED ONLY version above
-# 2. Uncomment the original LLM version below
-# ====================================================================

app/hybrid_industry_detector.py

@@ -1,28 +0,0 @@
-# app/hybrid_industry_detector.py
-import logging
-import pandas as pd
-from typing import Tuple, Dict
-from app.utils.detect_industry import detect_industry as rule_based_detect
-from app.service.ai_service import ai_service
-
-logger = logging.getLogger(__name__)
-
-def hybrid_detect_industry_type(org_id: str, df: pd.DataFrame, filename: str = "") -> Tuple[str, float, bool]:
-    """
-    Detects BUSINESS VERTICAL (SUPERMARKET/MANUFACTURING/PHARMA/RETAIL/WHOLESALE/HEALTHCARE)
-    
-    Returns: (industry, confidence, is_confident)
-    """
-    # 1. Rule-based detection from utils (<10ms, zero LLM cost)
-    industry, confidence = rule_based_detect(df)
-    industry = industry.upper()  # Normalize
-    
-    logger.info(f"[hybrid_industry] RULE-BASED ONLY: {industry} ({confidence:.2f})")
-    
-    # 2. [FUTURE] LLM fallback if confidence < 0.75
-    # if confidence < 0.75:
-    #     logger.info(f"[hybrid_industry] → LLM fallback needed")
-    #     # ... LLM logic here ...
-    
-    # 3. Always return as confident (rule-based is authoritative)
-    return industry, confidence, True

app/main.py

@@ -25,7 +25,7 @@ from app.core.worker_manager import worker_manager
 from app.deps import rate_limit_org, verify_api_key, check_all_services
 from app.tasks.analytics_worker import trigger_kpi_computation
 from app.service.vector_service import cleanup_expired_vectors
-from app.routers import health, datasources, reports, flags, scheduler, run, socket, analytics_stream,ai_query,schema
+from app.routers import health, datasources, reports, flags, scheduler, analytics_stream,ai_query,schema
 from app.service.llm_service import load_llm_service
 from app.deps import get_qstash_client
 from prometheus_client import make_asgi_app
@@ -422,8 +422,6 @@ app.include_router(datasources.router, prefix="/api/v1/datasources", dependencie
 app.include_router(reports.router, prefix="/api/v1/reports", dependencies=[Depends(verify_api_key)])
 app.include_router(flags.router, prefix="/api/v1/flags", dependencies=[Depends(verify_api_key)])
 app.include_router(scheduler.router, prefix="/api/v1/scheduler", dependencies=[Depends(verify_api_key)])
-app.include_router(run.router, prefix="/api/v1/run", dependencies=[Depends(verify_api_key)])
-app.include_router(socket.router, prefix="/api/v1/socket", dependencies=[Depends(verify_api_key)])
 app.include_router(analytics_stream.router, dependencies=[Depends(verify_api_key)])
 app.include_router(ai_query.router, prefix="/api/v1/ai-query", dependencies=[Depends(verify_api_key)])
 app.include_router(schema.router, prefix="/api/v1/schema", dependencies=[Depends(verify_api_key)])

app/mapper.py

@@ -22,7 +22,7 @@ import logging
 from typing import Dict, Any, Optional
 
 from app.db import get_conn, ensure_raw_table, transactional_conn, ensure_schema_versions_table
-from app.hybrid_entity_detector import hybrid_detect_entity_type
+from app.core.detection_engine import hybrid_detect_entity_type,hybrid_detect_industry_type
 from app.core.event_hub import event_hub
 from app.deps import get_sre_metrics
 from app.routers.health import emit_mapper_log
@@ -428,15 +428,15 @@ def _fallback_combined(org_id: str, source_id: str) -> tuple[dict, dict]:
             
             def detect_entity():
                 try:
-                    return hybrid_detect_entity_type(org_id, df, f"{source_id}.json")
+                    return hybrid_detect_entity_type(org_id, df, source_id, use_llm=False)
                 except Exception as e:
                     logger.error(f"[FALLBACK] Entity detection failed: {e}")
                     return ("UNKNOWN", 0.0, False)
             
             def detect_industry():
                 try:
-                    from app.hybrid_industry_detector import hybrid_detect_industry_type
-                    return hybrid_detect_industry_type(org_id, df, source_id)
+                    
+                    return hybrid_detect_industry_type(org_id, df, source_id, use_llm=False)
                 except Exception as e:
                     logger.error(f"[FALLBACK] Industry detection failed: {e}")
                     return ("UNKNOWN", 0.0, False)
@@ -528,8 +528,9 @@ def _fallback_industry_detection(org_id: str, source_id: str) -> dict:
         df = pd.DataFrame(parsed)
         df.columns = [str(col).lower().strip() for col in df.columns]
         
-        from app.hybrid_industry_detector import hybrid_detect_industry_type
-        industry, confidence, _ = hybrid_detect_industry_type(org_id, df, source_id)
+        from app.core.detection_engine import hybrid_detect_industry_type
+        industry, confidence, _ = hybrid_detect_industry_type(org_id, df, source_id, use_llm=False)
+
         
         industry_info = {"industry": industry, "confidence": confidence}
         logger.info(f"[FALLBACK_IND] ✅ Detected: {industry} ({confidence:.2%})")

app/routers/health.py

@@ -32,6 +32,7 @@ from app.tasks.analytics_worker import get_worker_manager
 from app.service.vector_service import VectorService
 from app.mapper import health_check_mapper, MapperMetrics
 from app.core.event_hub import  StreamingResponse, Response
+from app.core.sre_logging import log_aggregator, emit_worker_log, emit_vector_log, emit_llm_log, emit_mapper_log, emit_deps_log
 
 # Prometheus aggregation
 try:
@@ -52,68 +53,6 @@ except ImportError:
 logger = logging.getLogger(__name__)
 router = APIRouter(tags=["health"])
 
-# Global log aggregator (in-memory ring buffer for recent logs)
-class LogAggregator:
-    """Ring buffer storing last 1000 logs from all services"""
-    def __init__(self, max_size: int = 1000):
-        self.max_size = max_size
-        self.buffer: List[Dict[str, Any]] = []
-        self.lock = threading.Lock()
-    
-    def emit(self, service: str, level: str, message: str, **kwargs):
-        """Add a log entry from any service"""
-        with self.lock:
-            entry = {
-                "timestamp": datetime.utcnow().isoformat(),
-                "service": service,
-                "level": level,
-                "message": message,
-                **kwargs
-            }
-            self.buffer.append(entry)
-            if len(self.buffer) > self.max_size:
-                self.buffer.pop(0)
-    
-    def get_logs(self, service: Optional[str] = None, level: Optional[str] = None, limit: int = 100) -> List[Dict]:
-        """Retrieve filtered logs"""
-        with self.lock:
-            filtered = [
-                log for log in self.buffer
-                if (not service or log["service"] == service)
-                and (not level or log["level"] == level)
-            ]
-            return filtered[-limit:]
-
-    def get_error_rate(self, service: str, window_minutes: int = 5) -> float:
-        """Calculate error rate for a service"""
-        cutoff = datetime.utcnow() - timedelta(minutes=window_minutes)
-        recent = [
-            log for log in self.buffer
-            if log["service"] == service and log["timestamp"] >= cutoff.isoformat()
-        ]
-        if not recent:
-            return 0.0
-        errors = [log for log in recent if log["level"] in ("error", "critical")]
-        return len(errors) / len(recent)
-
-# Global aggregator instance
-log_aggregator = LogAggregator(max_size=1000)
-
-# Service-specific log emitters (to be imported by each service)
-def emit_worker_log(level: str, message: str, **kwargs):
-    log_aggregator.emit("analytics_worker", level, message, **kwargs)
-
-def emit_vector_log(level: str, message: str, **kwargs):
-    log_aggregator.emit("vector_service", level, message, **kwargs)
-
-def emit_llm_log(level: str, message: str, **kwargs):
-    log_aggregator.emit("llm_service", level, message, **kwargs)
-
-def emit_mapper_log(level: str, message: str, **kwargs):
-    log_aggregator.emit("mapper", level, message, **kwargs)
-
-def emit_deps_log(level: str, message: str, **kwargs):
-    log_aggregator.emit("dependencies", level, message, **kwargs)
 
 # ---------------------- SRE: Unified Health Endpoint ---------------------- #
 

app/routers/socket.py

@@ -1,54 +0,0 @@
-# app/routers/socket.py
-import socketio
-from fastapi import APIRouter, Depends, Path, Request
-from fastapi.responses import PlainTextResponse
-from app.deps import verify_api_key  # your API-key guard
-
-# 1️⃣ Socket.IO server
-sio = socketio.AsyncServer(
-    async_mode="asgi",
-    cors_allowed_origins=[
-        "https://mut-sync-hub.vercel.app",
-        "http://localhost:3000",
-    ],
-)
-
-# 2️⃣ ASGI sub-app (mounted separately in main.py)
-socket_app = socketio.ASGIApp(sio)
-
-# 3️⃣ FastAPI router for REST routes (no prefix → /socket-push)
-router = APIRouter(tags=["socket"])
-
-# ----------  POST /socket-push/{org_id} ----------
-@router.post("/socket-push/{org_id}")
-async def socket_push(
-    org_id: str = Path(...),
-    request: Request = None,
-    _: str = Depends(verify_api_key),
-):
-    """
-    Receive top-N rows from n8n workflow and broadcast them
-    live to all connected clients in the given org room.
-    """
-    payload = await request.json()
-    rows = payload.get("rows", [])
-    await sio.emit("datasource:new-rows", {"rows": rows}, room=org_id)
-    print(f"[socket] 🔄 broadcasted {len(rows)} rows → room={org_id}")
-    return {"status": "ok", "emitted": len(rows)}
-
-# ----------  Health Check ----------
-@router.get("/health")
-async def health():
-    return PlainTextResponse("ok")
-
-# ----------  Socket.IO Events ----------
-@sio.event
-async def connect(sid, environ, auth):
-    org_id = (auth or {}).get("orgId", "demo")
-    await sio.save_session(sid, {"orgId": org_id})
-    await sio.enter_room(sid, org_id)
-    print(f"[socket] ✅ {sid} connected → room={org_id}")
-
-@sio.event
-async def disconnect(sid):
-    print(f"[socket] ❌ {sid} disconnected")

app/service/ai_service.py

@@ -1,126 +0,0 @@
-import json
-import logging
-from app.deps import get_vector_db
-from typing import TYPE_CHECKING  # ✅ For forward reference
-import time
-
-if TYPE_CHECKING:
-    from app.service.llm_service import LocalLLMService  # ✅ Type hint only
-
-logger = logging.getLogger(__name__)
-
-class AIService:
-    def __init__(self):
-        try:
-            self.vector_db = get_vector_db()
-            self.vss_available = True
-            logger.info("✅ Vector DB initialized")
-        except Exception as e:
-            logger.warning(f"⚠️ Vector DB unavailable: {e}")
-            self.vector_db = None
-            self.vss_available = False
-        
-        self._llm = None  # ✅ Lazy initialization
-        self._embedder = None  # ✅ FIXED: Use _embedder (not embedder)
-        logger.info(f"✅ AI Service initialized (VSS: {'ENABLED' if self.vss_available else 'DISABLED'})")
-    
-    @property
-    def llm(self) -> "LocalLLMService":
-        """Lazy property to get LLM service (avoids circular import)"""
-        if self._llm is None:
-            from app.service.llm_service import get_llm_service  # ✅ Import INSIDE property
-            self._llm = get_llm_service()
-        return self._llm
-    
-    @property
-    def embedder(self):
-        """Lazy property to get embedder"""
-        if self._embedder is None:
-            from app.service.embedding_service import embedder  # ✅ Import INSIDE property
-            self._embedder = embedder
-        return self._embedder
-
-    
-    def detect_entity_type(self, org_id: str, columns: list[str], filename: str) -> dict:
-        """Detect entity type with JSON validation and timeout"""
-        columns_str = ",".join(columns)
-        
-        # Check cache
-        if self.vss_available:
-            cached = self.vector_db.execute("""
-                SELECT entity_type FROM vector_store.embeddings
-                WHERE org_id = ? AND content = ?
-                ORDER BY created_at DESC LIMIT 1
-            """, [org_id, columns_str]).fetchone()
-            
-            if cached:
-                logger.info(f"[ai_service] Cache hit: {cached[0]}")
-                return {"entity_type": cached[0], "confidence": 0.99, "cached": True}
-        
-        # ✅ SIMPLE, CLEAR prompt for Phi-3
-        prompt = f"""You are a data classification assistant. 
-        
-You MUST respond with ONLY valid JSON in this exact format:
-{{"entity_type":"sales|inventory|customer|product","confidence":0.95}}
-
-Dataset info:
-- Filename: {filename}
-- Columns: {columns_str}
-
-Analyze and respond with ONLY JSON:"""
-        
-        logger.info(f"[ai_service] Calling LLM for {org_id}...")
-        
-        try:
-            # ✅ TIMEOUT WRAPPER (5 seconds max)
-            start_time = time.time()
-            response_text = self.llm.generate(prompt, max_tokens=50, temperature=0.1)
-            elapsed = time.time() - start_time
-            
-            logger.info(f"[ai_service] LLM responded in {elapsed:.2f}s: {response_text}")
-            
-            # ✅ AGGRESSIVE JSON cleaning
-            response_text = response_text.strip()
-            if "```json" in response_text:
-                response_text = response_text.split("```json")[1].split("```")[0].strip()
-            elif "```" in response_text:
-                response_text = response_text.split("```")[1].split("```")[0].strip()
-            
-            # ✅ Remove any stray text before/after JSON
-            if "{" in response_text and "}" in response_text:
-                response_text = response_text[response_text.find("{"):response_text.rfind("}")+1]
-            
-            logger.info(f"[ai_service] Cleaned response: {response_text}")
-            
-            # ✅ PARSE with error handling
-            result = json.loads(response_text)
-            
-            # ✅ Normalize
-            result["entity_type"] = result["entity_type"].upper()
-            result["confidence"] = float(result["confidence"])
-            
-            # ✅ Cache it
-            if self.vss_available:
-                try:
-                    embedding = self.embedder.generate(columns_str)
-                    self.vector_db.execute("""
-                        INSERT INTO vector_store.embeddings (org_id, content, embedding, entity_type)
-                        VALUES (?, ?, ?, ?)
-                    """, [org_id, columns_str, embedding, result["entity_type"]])
-                    logger.info(f"[ai_service] Cached for {org_id}")
-                except Exception as e:
-                    logger.warning(f"[ai_service] Cache insert failed: {e}")
-            
-            return result
-            
-        except Exception as e:
-            logger.error(f"[ai_service] ❌ Detection failed: {e}", exc_info=True)
-            # ✅ SAFE FALLBACK (never crash)
-            return {
-                "entity_type": "SALES", 
-                "confidence": 0.50, 
-                "error": str(e),
-                "fallback": True
-            }
-
-ai_service = AIService()

app/service/llm_service.py

@@ -25,7 +25,7 @@ from typing import Optional, Dict, Any, List, Callable
 from dataclasses import dataclass, asdict
 import psutil  # For resource monitoring
 from fastapi import HTTPException
-from app.routers.health import emit_llm_log
+from app.core.sre_logging import emit_llm_log
 # Prometheus metrics (free tier compatible)
 try:
     from prometheus_client import Counter, Histogram, Gauge

app/service/vector_service.py

@@ -11,7 +11,7 @@ from sentence_transformers import SentenceTransformer
 import logging
 from datetime import datetime, timedelta
 from enum import Enum
-from app.routers.health import emit_vector_log
+from app.core.sre_logging import  emit_vector_log
 logger = logging.getLogger(__name__)
 
 

app/tasks/analytics_worker.py

@@ -26,7 +26,7 @@ from app.schemas.org_schema import OrgSchema
 from app.service.vector_service import VectorService, VectorStoreEventType, VectorMetrics
 from app.engine.kpi_calculators.registry import get_kpi_calculator_async
 from app.service.embedding_service import EmbeddingService
-from app.routers.health import emit_worker_log
+from app.core.sre_logging import emit_worker_log
 
 # Configure structured logging for SRE tools (Loki, etc.)
 logging.basicConfig(

app/tasks/worker.py

@@ -1,263 +1,397 @@
-# app/tasks/worker.py – ENTERPRISE GRADE (WITH ENTITY DETECTION)
+"""
+Worker v5.0: Pure LLM Detection Engine
+
+Purpose: Detect entity_type and industry using Phi-3 LLM
+- Queries DuckDB raw_rows for fresh data
+- Runs hybrid detection (LLM + rules)
+- Stores results in Redis for mapper to poll
+- Publishes pub/sub events for real-time subscribers
+- Zero legacy handlers, zero bloat
+
+SRE Features:
+- Structured JSON logging
+- Prometheus metrics per detection type
+- Circuit breaker for Redis failures
+- Request/response tracking with task_id
+- Error isolation and fallback to UNKNOWN
+"""
+
 import json
 import time
+import logging
 import signal
 import sys
 import traceback
 from typing import Dict, Any, Callable
-import pandas as pd  # ✅ Required for entity detection
+import pandas as pd
+import datetime
 
 from app.core.event_hub import event_hub
-from app.service.ai_service import ai_service
 from app.deps import get_duckdb
 from app.hybrid_entity_detector import hybrid_detect_entity_type, hybrid_detect_industry_type
+from app.core.sre_logging import emit_worker_log
 
-# ── Graceful Shutdown ──────────────────────────────────────────────────────────
+# ── SRE: Prometheus Metrics ─────────────────────────────────────────────────────
+try:
+    from prometheus_client import Counter, Histogram
+    detection_latency = Histogram(
+        'worker_detection_duration_seconds',
+        'Time to detect entity/industry',
+        ['detection_type', 'org_id']
+    )
+    detection_errors = Counter(
+        'worker_detection_errors_total',
+        'Total detection failures',
+        ['detection_type', 'org_id', 'error_type']
+    )
+except ImportError:
+    detection_latency = None
+    detection_errors = None
+
+# ── Logging Setup ───────────────────────────────────────────────────────────────
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s | [%(levelname)s] [%(name)s] %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+# ── Graceful Shutdown ───────────────────────────────────────────────────────────
 def shutdown(signum, frame):
-    print("\n🛑 Worker shutting down gracefully...")
+    logger.info("🛑 Worker shutting down gracefully...")
     sys.exit(0)
 
 signal.signal(signal.SIGINT, shutdown)
 signal.signal(signal.SIGTERM, shutdown)
 
+# ── CORE: LLM-Based Detection Handlers ──────────────────────────────────────────
 
-
-# ── NEW: Entity Detection Handler ───────────────────────────────────────────────
-def process_detect_entity(org_id: str, **args):
-    """🎯 Queries DuckDB for raw data instead of receiving payload"""
+def process_detect_entity(org_id: str, **args) -> Dict[str, Any]:
+    """
+    🎯 MAIN: Detect entity_type using LLM queries to DuckDB
+    
+    Flow:
+    1. Query latest raw rows from DuckDB
+    2. Run hybrid LLM detection (Phi-3 + rules)
+    3. Store result in Redis (mapper polls this)
+    4. Publish pub/sub event for real-time subscribers
+    5. Return structured result
+    
+    Args:
+        org_id: Organization ID
+        source_id: From args["source_id"]
+        
+    Returns:
+        {"entity_type": str, "confidence": float, "source_id": str, "status": str}
+    """
+    start_time = time.time()
     source_id = args["source_id"]
+    task_id = args.get("task_id", "unknown")
     
-    print(f"🔵 [{org_id}] Entity detection STARTED for {source_id}")
+    emit_worker_log("info", "Entity detection started", 
+                   org_id=org_id, source_id=source_id, task_id=task_id)
     
     try:
-        # 1. ✅ Query DuckDB for latest raw rows (the ones just uploaded)
+        # 1. Query DuckDB for raw data (the data just uploaded)
         conn = get_duckdb(org_id)
         rows = conn.execute("""
-             SELECT row_data
+            SELECT row_data
             FROM main.raw_rows
             WHERE row_data IS NOT NULL
             USING SAMPLE 40
         """).fetchall()
-
         
         if not rows:
             raise RuntimeError(f"No raw data found for {source_id}")
         
-        # 2. Parse into DataFrame
+        # 2. Parse to DataFrame for LLM detection
         parsed = [json.loads(r[0]) for r in rows if r[0]]
         df = pd.DataFrame(parsed)
-        print(f"   📊 DataFrame: {len(df)} rows × {len(df.columns)} cols")
+        logger.info(f"[WORKER] 📊 Entity detection DataFrame: {len(df)} rows × {len(df.columns)} cols")
         
-        # 3. Detect entity (rest unchanged)
-        entity_type, confidence, _ = hybrid_detect_entity_type(org_id, df, f"{source_id}.json")
-        print(f"   ✅ Detected: {entity_type} ({confidence:.2%})")
+        # 3. Run hybrid LLM detection (Phi-3 + rules)
+        entity_type, confidence, _ = hybrid_detect_entity_type(org_id, df, source_id, use_llm=True)
+        logger.info(f"[WORKER] ✅ Entity detected: {entity_type} ({confidence:.2%})")
         
-        # 3. Store in Redis for mapper to poll (HF endpoint is waiting for this)
+        # 4. Store in Redis (mapper's poll_for_entity() reads this)
         entity_key = f"entity:{org_id}:{source_id}"
-        event_hub.setex(
-            entity_key,
-            3600,  # 1 hour TTL (gives mapper plenty of time)
-            json.dumps({
-                "entity_type": entity_type,
-                "confidence": confidence,
-                "detected_at": time.time(),
-                "source_id": source_id
-            })
-        )
-        print(f"   💾 Stored in Redis: {entity_key}")
+        entity_data = {
+            "entity_type": entity_type,
+            "confidence": confidence,
+            "detected_at": time.time(),
+            "source_id": source_id,
+            "detected_by": "llm-worker"
+        }
+        
+        event_hub.setex(entity_key, 3600, json.dumps(entity_data))
+        emit_worker_log("info", "Entity stored in Redis", 
+                       org_id=org_id, source_id=source_id, entity_type=entity_type)
         
-        # 4. Publish event for any real-time subscribers (future-proofing)
+        # 5. Publish pub/sub event for real-time subscribers
         event_hub.publish(
             f"entity_ready:{org_id}",
             json.dumps({
                 "source_id": source_id,
                 "entity_type": entity_type,
-                "confidence": confidence
+                "confidence": confidence,
+                "timestamp": datetime.utcnow().isoformat()
             })
         )
-        print(f"   📤 Published to entity_ready:{org_id}")
+        emit_worker_log("debug", "Pub/sub event published", channel=f"entity_ready:{org_id}")
         
-        # 5. Return result to satisfy worker's response contract
+        # 6. SRE: Record metrics
+        if detection_latency:
+            detection_latency.labels(detection_type="entity", org_id=org_id).observe(
+                (time.time() - start_time)
+            )
+        
+        # 7. Return structured result
         return {
             "entity_type": entity_type,
             "confidence": confidence,
             "source_id": source_id,
-            "status": "stored_in_redis"
+            "status": "stored_in_redis",
+            "task_id": task_id,
+            "duration_ms": round((time.time() - start_time) * 1000, 2)
         }
         
     except Exception as e:
-        print(f"❌ Entity detection failed: {e}")
-        # CRITICAL: Re-raise so process_task logs it properly
-        raise RuntimeError(f"Entity detection failed for {source_id}: {str(e)}")
+        error_msg = f"Entity detection failed for {source_id}: {str(e)}"
+        logger.error(f"[WORKER] {error_msg}")
+        
+        # SRE: Record error
+        if detection_errors:
+            detection_errors.labels(detection_type="entity", org_id=org_id, error_type=type(e).__name__).inc()
+        
+        emit_worker_log("error", "Entity detection failed", 
+                       org_id=org_id, source_id=source_id, error=error_msg)
+        
+        # Fallback: Store UNKNOWN to unblock mapper
+        event_hub.setex(f"entity:{org_id}:{source_id}", 3600, json.dumps({
+            "entity_type": "UNKNOWN",
+            "confidence": 0.0,
+            "detected_at": time.time(),
+            "source_id": source_id,
+            "error": error_msg
+        }))
+        
+        raise RuntimeError(error_msg)
 
-def process_detect_industry(org_id: str, **args):
+def process_detect_industry(org_id: str, **args) -> Dict[str, Any]:
     """
-    🎯 DETECTS INDUSTRY (business vertical) only.
-    DOES NOT touch entity detection.
+    🎯 MAIN: Detect industry vertical using LLM
+    
+    Flow:
+    1. Query DuckDB raw rows
+    2. Run hybrid LLM detection
+    3. Store result in Redis
+    4. Publish pub/sub event
+    5. Also triggers entity detection (independent task)
+    
+    Args:
+        org_id: Organization ID
+        source_id: From args["source_id"]
+        
+    Returns:
+        {"industry": str, "confidence": float, "source_id": str, "status": str}
     """
+    start_time = time.time()
     source_id = args["source_id"]
+    task_id = args.get("task_id", "unknown")
     
-    print(f"🔴🔴🔴 [WORKER] INDUSTRY detection for {org_id}/{source_id}")
+    emit_worker_log("info", "Industry detection started", 
+                   org_id=org_id, source_id=source_id, task_id=task_id)
     
     try:
-        # Query raw data
+        # 1. Query DuckDB
         conn = get_duckdb(org_id)
-        rows = conn.execute("SELECT row_data FROM main.raw_rows LIMIT 100").fetchall()
+        rows = conn.execute("""
+            SELECT row_data
+            FROM main.raw_rows
+            WHERE row_data IS NOT NULL
+            USING SAMPLE 40
+        """).fetchall()
         
         if not rows:
-            raise RuntimeError("No raw data")
+            raise RuntimeError(f"No raw data found for {source_id}")
         
+        # 2. Parse DataFrame
         parsed = [json.loads(r[0]) for r in rows if r[0]]
         df = pd.DataFrame(parsed)
+        logger.info(f"[WORKER] 📊 Industry detection DataFrame: {len(df)} rows × {len(df.columns)} cols")
         
-        # ✅ Use NEW detector (decoupled from entity)
-        industry, confidence, _ = hybrid_detect_industry_type(org_id, df, source_id)
+        # 3. Run hybrid LLM detection
+        industry, confidence, _ = hybrid_detect_industry_type(org_id, df, source_id, use_llm=True)
+        logger.info(f"[WORKER] ✅ Industry detected: {industry} ({confidence:.2%})")
         
-        # Write to dedicated Redis key
-        event_hub.setex(f"industry:{org_id}:{source_id}", 3600, json.dumps({
+        # 4. Store in Redis
+        industry_key = f"industry:{org_id}:{source_id}"
+        industry_data = {
             "industry": industry,
-            "confidence": confidence
-        }))
+            "confidence": confidence,
+            "detected_at": time.time(),
+            "source_id": source_id,
+            "detected_by": "llm-worker"
+        }
         
-        print(f"✅ [WORKER] INDUSTRY written: {industry} ({confidence:.2%})")
+        event_hub.setex(industry_key, 3600, json.dumps(industry_data))
+        emit_worker_log("info", "Industry stored in Redis", 
+                       org_id=org_id, source_id=source_id, industry=industry)
         
-        # Auto-queue entity detection (separate task, independent)
+        # 5. Publish pub/sub event
+        event_hub.publish(
+            f"industry_ready:{org_id}",
+            json.dumps({
+                "source_id": source_id,
+                "industry": industry,
+                "confidence": confidence,
+                "timestamp": datetime.utcnow().isoformat()
+            })
+        )
+        
+        # 6. Auto-trigger entity detection (independent task)
+        # This ensures both entity and industry are eventually detected
         entity_task = {
             "id": f"detect_entity:{org_id}:{source_id}:{int(time.time())}",
             "function": "detect_entity",
             "args": {"org_id": org_id, "source_id": source_id}
         }
         event_hub.lpush("python:task_queue", json.dumps(entity_task))
+        emit_worker_log("debug", "Auto-triggered entity detection", 
+                       org_id=org_id, source_id=source_id)
+        
+        # 7. SRE: Record metrics
+        if detection_latency:
+            detection_latency.labels(detection_type="industry", org_id=org_id).observe(
+                (time.time() - start_time)
+            )
+        
+        return {
+            "industry": industry,
+            "confidence": confidence,
+            "source_id": source_id,
+            "status": "stored_in_redis",
+            "task_id": task_id,
+            "duration_ms": round((time.time() - start_time) * 1000, 2)
+        }
         
     except Exception as e:
-        print(f"❌ [WORKER] Industry detection CRASHED: {e}")
+        error_msg = f"Industry detection failed for {source_id}: {str(e)}"
+        logger.error(f"[WORKER] {error_msg}")
+        
+        if detection_errors:
+            detection_errors.labels(detection_type="industry", org_id=org_id, error_type=type(e).__name__).inc()
+        
+        emit_worker_log("error", "Industry detection failed", 
+                       org_id=org_id, source_id=source_id, error=error_msg)
+        
+        # Fallback: Store UNKNOWN
         event_hub.setex(f"industry:{org_id}:{source_id}", 3600, json.dumps({
             "industry": "UNKNOWN",
-            "confidence": 0.0
+            "confidence": 0.0,
+            "detected_at": time.time(),
+            "source_id": source_id,
+            "error": error_msg
         }))
-        raise
-
-# ── Legacy Handlers (Keep for backward compatibility) ────────────────────────
-def canonify_df_with_entity(org_id: str, filename: str, hours_window: int = 24):
-    """⚠️ DEPRECATED: Remove once all ingestion uses detect_entity worker"""
-    from app.mapper import canonify_df
-    return canonify_df(org_id, filename, hours_window)
+        
+        raise RuntimeError(error_msg)
 
-def execute_org_sql(org_id: str, sql: str):
-    """Execute SQL for specific org with enterprise guardrails"""
-    conn = get_duckdb(org_id)
-    
-    # 🔒 Security: Whitelist only SELECT queries
-    safe_sql = sql.strip().upper()
-    if not safe_sql.startswith("SELECT"):
-        raise ValueError("🔒 Only SELECT queries allowed")
-    
-    # 💡 Safety: Auto-limit to prevent memory overload
-    if "LIMIT" not in safe_sql:
-        safe_sql += " LIMIT 10000"
-    
-    return conn.execute(safe_sql).fetchall()
+# ── Task Registry (CLEAN – Only LLM Detection) ──────────────────────────────────
 
-# ── Task Handler Registry ─────────────────────────────────────────────────────
-# ⚠️ ORDER MATTERS: Add new handlers at the top for visibility
 TASK_HANDLERS: Dict[str, Callable] = {
-    "detect_entity": process_detect_entity,  # 🎯 NEW: Ingestion-critical path
-    
-    # Legacy AI handlers (keep until fully migrated)
-    "detect_entity_type": lambda org_id, **args: ai_service.detect_entity_type(org_id, **args),
-    "generate_sql": lambda org_id, **args: ai_service.generate_sql(org_id, **args),
-    "generate_insights": lambda org_id, **args: ai_service.generate_insights(org_id, **args),
-    "similarity_search": lambda org_id, **args: ai_service.similarity_search(org_id, **args),
-    
-    # Legacy mapper handlers (to be deprecated)
-    "canonify_df": canonify_df_with_entity,
-    "execute_sql": execute_org_sql,
+    "detect_entity": process_detect_entity,      # 🎯 LLM entity detection
+    "detect_industry": process_detect_industry,  # 🎯 LLM industry detection
+    # ✅ All legacy handlers removed – mapper handles the rest via polling
 }
 
-# ── Task Processing (UNCHANGED – BATTLE TESTED) ───────────────────────────────
-def process_task(task_data: Dict[str, Any]):
-    """Process single task with full observability and error isolation"""
-    task_id = task_data.get("id")
+# ── Task Processing (SIMPLIFIED – No Legacy) ────────────────────────────────────
+
+def process_task(task_data: Dict[str, Any]) -> None:
+    """
+    Process single detection task with SRE observability
+    
+    Args:
+        task_data: {"id": str, "function": str, "args": dict}
+    """
+    start_time = time.time()
+    task_id = task_data.get("id", "unknown")
     function_name = task_data.get("function")
     args = task_data.get("args", {})
     
-    # ── Validation ─────────────────────────────────────────────────────────────
-    if not task_id or not function_name:
-        raise ValueError("❌ Invalid task: missing id or function")
-    
-    if "org_id" not in args:
-        raise ValueError(f"❌ Task {task_id} missing required org_id")
-    
-    org_id = args["org_id"]
+    org_id = args.get("org_id", "unknown")
+    source_id = args.get("source_id", "unknown")
     
-    # ── Execution ──────────────────────────────────────────────────────────────
-    start_time = time.time()
-    print(f"🔵 [{org_id}] Processing {function_name} (task: {task_id})")
+    emit_worker_log("info", "Task processing started", 
+                   task_id=task_id, function=function_name, org_id=org_id, source_id=source_id)
     
     try:
         handler = TASK_HANDLERS.get(function_name)
         if not handler:
-            raise ValueError(f"❌ Unknown function: {function_name}")
+            raise ValueError(f"Unknown detection function: {function_name}")
         
         # Execute handler
         result = handler(org_id, **args)
         
-        # ── Success ────────────────────────────────────────────────────────────
         duration = time.time() - start_time
-        print(f"✅ [{org_id}] {function_name} completed in {duration:.2f}s")
         
-        event_hub.setex(
-            f"python:response:{task_id}",
-            3600,
-            json.dumps({
-                "status": "success",
-                "org_id": org_id,
-                "function": function_name,
-                "data": result,
-                "duration": duration
-            })
-        )
+        # Store success response
+        response_key = f"python:response:{task_id}"
+        event_hub.setex(response_key, 3600, json.dumps({
+            "status": "success",
+            "function": function_name,
+            "org_id": org_id,
+            "data": result,
+            "duration": duration
+        }))
+        
+        emit_worker_log("info", "Task completed", 
+                       task_id=task_id, function=function_name, 
+                       duration_ms=round(duration * 1000, 2))
         
     except Exception as e:
-        # ── Error ──────────────────────────────────────────────────────────────
         duration = time.time() - start_time
-        error_msg = f"{type(e).__name__}: {str(e)}"
-        print(f"❌ [{org_id}] {function_name} FAILED after {duration:.2f}s: {error_msg}")
-        print(traceback.format_exc())  # Full trace for debugging
+        error_type = type(e).__name__
         
-        event_hub.setex(
-            f"python:response:{task_id}",
-            3600,
-            json.dumps({
-                "status": "error",
-                "org_id": org_id,
-                "function": function_name,
-                "message": error_msg,
-                "duration": duration
-            })
-        )
+        # Store error response
+        response_key = f"python:response:{task_id}"
+        event_hub.setex(response_key, 3600, json.dumps({
+            "status": "error",
+            "function": function_name,
+            "org_id": org_id,
+            "message": str(e),
+            "duration": duration
+        }))
+        
+        emit_worker_log("error", "Task failed", 
+                       task_id=task_id, function=function_name, 
+                       error=str(e), error_type=error_type)
+        
+        # Re-raise to let caller know
+        raise
+
+# ── Main Worker Loop (UNCHANGED – BATTLE TESTED) ───────────────────────────────
 
-# ── Main Worker Loop (UNCHANGED – HANDLES MILLIONS OF TASKS) ──────────────────
 if __name__ == "__main__":
-    print("🚀 Python worker listening on Redis queue...")
-    print("Press Ctrl+C to stop")
+    logger.info("🚀 Python detection worker listening on Redis queue...")
+    logger.info("Press Ctrl+C to stop")
     
     while True:
         try:
             # Blocking pop (0 = infinite wait, no CPU burn)
-            _, task_json = event_hub.brpop("python:task_queue", timeout=0)
+            result = event_hub.brpop("python:task_queue", timeout=0)
             
-            try:
-                task_data = json.loads(task_json)
-                process_task(task_data)
-            except json.JSONDecodeError as e:
-                print(f"❌ Malformed task JSON: {e}")
-                continue
+            if result:
+                _, task_json = result
+                try:
+                    task_data = json.loads(task_json)
+                    process_task(task_data)
+                except json.JSONDecodeError as e:
+                    logger.error(f"Malformed task JSON: {e}")
+                    continue
             
         except KeyboardInterrupt:
-            print("\nShutting down...")
+            logger.info("Shutting down...")
             break
         except Exception as e:
-            print(f"🔴 WORKER-LEVEL ERROR (will restart): {e}")
+            logger.error(f"🔴 WORKER-LEVEL ERROR (will restart): {e}")
             traceback.print_exc()
             time.sleep(5)  # Cooldown before retry