updated backend

cluster_orchestrator.py

@@ -186,13 +186,15 @@ class ClusterOrchestrator:
             # Use ALL messages in the conversation for context
             conversation_texts = [messages[i].text for i in message_indices]
             
-            # Generate conversation label (shorter, more specific)
-            if len(conversation_texts) <= 3:
-                # For short conversations, use the first message as label
+            # Always try to use LLM for better labels if we have enough content
+            # Only fall back to simple truncation for extremely short/empty convos
+            total_chars = sum(len(t) for t in conversation_texts)
+            
+            if total_chars < 50:
+                # Very short conversation: use first message
                 first_msg = conversation_texts[0]
                 if len(first_msg) > 60:
                     truncated = first_msg[:60]
-                    # Try to break at word boundary
                     last_space = truncated.rfind(' ')
                     if last_space > 0:
                         truncated = truncated[:last_space]
@@ -200,8 +202,13 @@ class ClusterOrchestrator:
                 else:
                     label = first_msg
             else:
-                # For longer conversations, generate a summary label
-                label = self.label_service.generate_cluster_label(conversation_texts[:10])
+                # Use Gemini for proper labeling of the conversation
+                # This fixes the issue of "I'll Have Let's" type labels
+                label = self.label_service.generate_cluster_label(
+                    conversation_texts, 
+                    max_messages=10, # Fewer messages needed for single conversation
+                    max_length=40    # Shorter labels for leaf nodes
+                )
             
             # Get channel info (store in metadata, not in label)
             channel = messages[message_indices[0]].channel if message_indices else "unknown"

gemini_label_service.py

@@ -48,30 +48,33 @@ class GeminiLabelService:
     def generate_cluster_label(
         self,
         messages: List[str],
-        max_messages: int = 10,
-        max_length: int = 50
+        max_messages: int = 30,
+        max_length: int = 60
     ) -> str:
         """Generate a descriptive label for a cluster"""
         if not messages:
             return "Empty Cluster"
         
         selected = messages[:max_messages]
-        messages_text = "\n".join([f"- {msg[:150]}" for msg in selected])
+        # Allow slightly longer context per message
+        messages_text = "\n".join([f"- {msg[:200]}" for msg in selected])
         
-        prompt = f"""Analyze these chat messages and create a clear, descriptive topic label in 3-6 words.
-Be specific and concise.
+        prompt = f"""Analyze these chat messages from a team collaboration channel.
+Identify the main project, specific technical issue, or key activity being discussed.
+Create a descriptive, specific title (4-8 words) that clearly distinguishes this topic.
+Avoid generic phrases like "Team Discussion" or "Project Update".
 
 Messages:
 {messages_text}
 
-Topic label (3-6 words):"""
+Specific Topic Title:"""
         
         try:
             response = self.model.generate_content(
                 prompt,
                 generation_config=genai.types.GenerationConfig(
-                    max_output_tokens=20,
-                    temperature=0.7,
+                    max_output_tokens=30,
+                    temperature=0.4,  # Lower temperature for more focused results
                 )
             )
             
@@ -124,10 +127,20 @@ Keywords:"""
     
     def _clean_label(self, label: str) -> str:
         """Clean and format label"""
-        label = label.replace("Topic:", "").replace("topic:", "").strip()
+        # Remove common prefixes/suffixes from LLM output
+        prefixes = ["Title:", "Label:", "Topic:", "Subject:", "The topic is", "Discussion about"]
+        for prefix in prefixes:
+            if label.lower().startswith(prefix.lower()):
+                label = label[len(prefix):].strip()
+        
+        # Remove quotes if present
+        label = label.strip('"\'')
+        
+        # Capitalize first letter
         if label and not label[0].isupper():
             label = label[0].upper() + label[1:]
-        return label[:50] if label else "General Discussion"
+            
+        return label[:60] if label else "General Discussion"
     
     def _clean_tag(self, tag: str) -> str:
         """Clean a tag"""
@@ -136,19 +149,38 @@ Keywords:"""
     
     def _fallback_label(self, messages: List[str]) -> str:
         """Simple fallback if API fails"""
+        if not messages:
+            return "General Discussion"
+        
+        # Try to use the beginning of the first substantial message
+        for msg in messages:
+            if len(msg) > 20:
+                # Find first sentence or up to 50 chars
+                end = msg.find('.')
+                if end > 0:
+                    candidate = msg[:end+1]
+                else:
+                    candidate = msg
+                
+                if len(candidate) > 60:
+                    candidate = candidate[:60].rsplit(' ', 1)[0] + "..."
+                
+                return candidate
+        
+        # Fallback to word counter if all messages are tiny
         from collections import Counter
         words = []
         for msg in messages:
             words.extend(msg.lower().split())
         
-        stopwords = {"the", "a", "an", "and", "or", "but", "in", "on", "at", "to", "for"}
+        stopwords = {"the", "a", "an", "and", "or", "but", "in", "on", "at", "to", "for", "is", "are", "of", "with"}
         words = [w for w in words if w not in stopwords and len(w) > 3]
         
         if not words:
             return "General Discussion"
         
-        common = Counter(words).most_common(3)
-        return " ".join([word.capitalize() for word, _ in common])
+        common = Counter(words).most_common(2)
+        return " & ".join([word.capitalize() for word, _ in common])
     
     def _fallback_tags(self, messages: List[str], num_tags: int) -> List[str]:
         """Simple fallback tags"""

hierarchical_clustering_service.py

@@ -107,7 +107,11 @@ class HierarchicalClusteringService:
         
         # Cluster conversations by semantic similarity
         if len(conversations) > 1:
-            topic_labels = self._create_topic_clusters(conversation_embeddings, len(conversations))
+            topic_labels = self._create_topic_clusters(
+                conversation_embeddings, 
+                len(conversations),
+                n_messages
+            )
         else:
             topic_labels = np.array([0])
         
@@ -160,37 +164,45 @@ class HierarchicalClusteringService:
     def _create_topic_clusters(
         self,
         conversation_embeddings: np.ndarray,
-        n_conversations: int
+        n_conversations: int,
+        n_messages: int
     ) -> np.ndarray:
         """
         Cluster conversations by topic using semantic similarity.
         Returns cluster labels for each conversation as numpy array.
         
-        Note: main_cluster_threshold (default 1.2) controls topic granularity.
-        Lower values = fewer, broader topics; higher values = more, specific topics.
+        Uses dynamic cluster counting: ~5% of total messages, capped by max_clusters.
         """
         if n_conversations < 2:
             return np.array([0])
         
-        # First, try with threshold-based clustering
-        labels = self._cluster_level(
-            conversation_embeddings,
-            self.main_cluster_threshold,
-            self.min_main_cluster_size
-        )
+        # Calculate target number of clusters based on message count (5% rule)
+        # Example: 100 messages -> 5 clusters
+        target_n_clusters = int(n_messages * 0.05)
+        
+        # Ensure reasonable bounds
+        # At least 2 clusters (if we have enough conversations)
+        # At most max_clusters
+        # At most n_conversations (can't have more clusters than items)
+        target_n_clusters = max(2, target_n_clusters)
+        target_n_clusters = min(target_n_clusters, self.max_clusters)
+        target_n_clusters = min(target_n_clusters, n_conversations)
         
-        n_clusters = len(np.unique(labels))
+        logger.info(f"Clustering {n_conversations} conversations into {target_n_clusters} topics "
+                   f"(based on {n_messages} messages)")
         
-        # If we have too many clusters, use n_clusters parameter instead
-        if n_clusters > self.max_clusters:
-            logger.info(f"Threshold produced {n_clusters} topic clusters, limiting to {self.max_clusters}")
+        try:
             clustering = AgglomerativeClustering(
-                n_clusters=self.max_clusters,
-                linkage='ward'
+                n_clusters=target_n_clusters,
+                linkage='ward',
+                metric='euclidean'
             )
             labels = clustering.fit_predict(conversation_embeddings)
-        
-        return labels
+            return labels
+        except Exception as e:
+            logger.error(f"Topic clustering failed: {e}")
+            # Fallback to single cluster
+            return np.zeros(n_conversations, dtype=int)
     
     def _cluster_level(
         self,

main.py

@@ -5,6 +5,7 @@ from fastapi import FastAPI, HTTPException, BackgroundTasks
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from typing import Dict, Optional
+from supabase_job_storage import get_job_storage
 import logging
 import uuid
 
@@ -49,8 +50,9 @@ app.include_router(slack_oauth_router)
 app.include_router(discord_oauth_router)
 
 # Job storage (in production, use Redis or database)
-jobs: Dict[str, ClusteringStatus] = {}
-results: Dict[str, ClusteringOutput] = {}
+#jobs: Dict[str, ClusteringStatus] = {}
+#results: Dict[str, ClusteringOutput] = {}
+storage = get_job_storage()
 
 @app.on_event("startup")
 async def startup_event():
@@ -72,6 +74,7 @@ async def startup_event():
         logger.warning(f"Warmup failed ({type(e).__name__}): {e}")
     except Exception as e:
         logger.warning(f"Warmup failed with unexpected error ({type(e).__name__}): {e}")
+    storage.cleanup_old_jobs()
 
 @app.get("/")
 async def root():
@@ -145,11 +148,20 @@ async def cluster_messages_async(
     job_id = str(uuid.uuid4())
     
     # Initialize job status
-    jobs[job_id] = ClusteringStatus(
+    # jobs[job_id] = ClusteringStatus(
+    #     status="processing",
+    #     progress=0.0,
+    #     message="Starting clustering job",
+    #     job_id=job_id
+    # )
+    storage.create_job(
+        job_id=job_id,
         status="processing",
         progress=0.0,
         message="Starting clustering job",
-        job_id=job_id
+        distance_threshold=request.distance_threshold,
+        min_cluster_size=request.min_cluster_size,
+        force_recluster=request.force_recluster
     )
     
     # Add background task
@@ -165,13 +177,23 @@ async def cluster_messages_async(
 async def process_clustering_job(job_id: str, request: ClusteringRequest):
     """Background task for clustering"""
     try:
-        jobs[job_id].message = "Processing messages..."
-        jobs[job_id].progress = 10.0
+        # jobs[job_id].message = "Processing messages..."
+        # jobs[job_id].progress = 10.0
+        storage.update_job_status(
+            job_id=job_id,
+            message="Processing messages...",
+            progress=10.0
+        )
         
         orchestrator = get_orchestrator()
         
-        jobs[job_id].message = "Generating embeddings..."
-        jobs[job_id].progress = 30.0
+        # jobs[job_id].message = "Generating embeddings..."
+        # jobs[job_id].progress = 30.0
+        storage.update_job_status(
+            job_id=job_id,
+            message="Generating embeddings...",
+            progress=30.0
+        )
         
         result = orchestrator.process_messages(
             messages=request.messages,
@@ -180,43 +202,62 @@ async def process_clustering_job(job_id: str, request: ClusteringRequest):
             min_cluster_size=request.min_cluster_size
         )
         
-        jobs[job_id].message = "Clustering complete"
-        jobs[job_id].progress = 100.0
-        jobs[job_id].status = "completed"
-        
-        results[job_id] = result
+        # jobs[job_id].message = "Clustering complete"
+        # jobs[job_id].progress = 100.0
+        #jobs[job_id].status = "completed"
+        #results[job_id] = result
+        storage.save_result(job_id, result.dict())
+        storage.update_job_status(
+            job_id=job_id,
+            message="Clustering complete",
+            progress=100.0,
+            status="completed"
+        )
         
     except Exception as e:
         logger.error(f"Error in background job {job_id}: {e}", exc_info=True)
-        jobs[job_id].status = "error"
-        jobs[job_id].message = str(e)
+        # jobs[job_id].status = "error"
+        # jobs[job_id].message = str(e)
+        storage.update_job_status(
+            job_id=job_id,
+            message=str(e),
+            status="error"
+        )
 
 
 @app.get("/cluster/status/{job_id}", response_model=ClusteringStatus)
 async def get_job_status(job_id: str):
     """Get status of a clustering job"""
-    if job_id not in jobs:
-        raise HTTPException(status_code=404, detail="Job not found")
-    
-    return jobs[job_id]
+    # if job_id not in jobs:    
+    # return jobs[job_id]
+    job_data = storage.get_job(job_id)
+    if not job_data:
+        raise HTTPException(status_code=404, detail="Job not found or expired")
+    return ClusteringStatus(**job_data)
 
 
 @app.get("/cluster/result/{job_id}", response_model=ClusteringOutput)
 async def get_job_result(job_id: str):
     """Get result of a completed clustering job"""
-    if job_id not in jobs:
-        raise HTTPException(status_code=404, detail="Job not found")
+    # if job_id not in jobs:
+    job_data = storage.get_job(job_id)
+    if not job_data:
+        raise HTTPException(status_code=404, detail="Job not found or expired")
     
-    if jobs[job_id].status != "completed":
+    # if jobs[job_id].status != "completed":
+    if job_data["status"] != "completed":
         raise HTTPException(
             status_code=400,
-            detail=f"Job not completed. Current status: {jobs[job_id].status}"
+            detail=f"Job not completed. Current status: {job_data['status']}"
         )
     
-    if job_id not in results:
-        raise HTTPException(status_code=404, detail="Result not found")
-    
-    return results[job_id]
+    # if job_id not in results:
+    #    raise HTTPException(status_code=404, detail="Result not found")
+    # return results[job_id]
+    result_data = storage.get_result(job_id)
+    if not result_data:
+        raise HTTPException(status_code=404, detail="Result not found or expired")
+    return ClusteringOutput(**result_data)
 
 
 @app.post("/search", response_model=list[SearchResult])
@@ -231,12 +272,48 @@ async def search_messages(request: SearchRequest):
         List of search results
     """
     try:
-        # This endpoint requires messages to be provided or stored
-        # For now, return error - in production, integrate with database
-        raise HTTPException(
-            status_code=501,
-            detail="Search endpoint requires integration with message storage"
+        orchestrator = get_orchestrator()
+        
+        # If messages are provided in request (not ideal but works for small batches)
+        # In a real app, we'd use a job_id or session_id to retrieve stored messages
+        if not request.messages_with_tags:
+             # Fallback: check if we have a recent result in memory (simple stateful approach)
+             # This is a hack for the demo; in prod use a DB
+             
+            #  if results:
+            #      last_job_id = list(results.keys())[-1]
+            #      request.messages_with_tags = results[last_job_id].messages
+             recent_jobs = storage.get_recent_jobs(limit=1)
+             if recent_jobs:
+                result_data = storage.get_result(recent_jobs[0]["job_id"])
+                if result_data:
+                    request.messages_with_tags = [
+                        MessageWithTags(**msg) for msg in result_data.get("messages", [])
+                    ]
+             else:
+                raise HTTPException(
+                    status_code=400, 
+                    detail="No context provided for search. Please run clustering first."
+                )
+
+        results_tuples = orchestrator.search_messages(
+            query=request.query,
+            messages_with_tags=request.messages_with_tags,
+            filter_tags=request.filter_tags,
+            filter_clusters=request.filter_clusters,
+            top_k=request.top_k
         )
+        
+        # Convert tuples to SearchResult objects
+        search_results = [
+            SearchResult(
+                message=msg,
+                score=score
+            )
+            for msg, score in results_tuples
+        ]
+        
+        return search_results
     
     except HTTPException:
         raise
@@ -363,6 +440,17 @@ async def fetch_slack_messages(request: SlackFetchRequest):
         raise HTTPException(status_code=500, detail=str(e))
 
 
+@app.post("/admin/cleanup-jobs")
+async def cleanup_old_jobs():
+    """Manually trigger cleanup of old jobs (>48 hours)"""
+    deleted_count = storage.cleanup_old_jobs()
+    return {
+        "status": "success",
+        "deleted_jobs": deleted_count,
+        "message": f"Cleaned up {deleted_count} old jobs"
+    }
+
+
 if __name__ == "__main__":
     import uvicorn
     

models.py

@@ -73,12 +73,15 @@ class SearchRequest(BaseModel):
     top_k: int = Field(10, description="Number of results to return")
     filter_tags: Optional[List[str]] = Field(None, description="Filter by specific tags")
     filter_clusters: Optional[List[str]] = Field(None, description="Filter by specific clusters")
+    # Add optional context messages for the search
+    messages_with_tags: Optional[List[MessageWithTags]] = Field(None, description="Context messages to search within")
 
 class SearchResult(BaseModel):
     """Single search result"""
     message: MessageWithTags
-    similarity_score: float = Field(..., description="Similarity score to query")
-    rank: int = Field(..., description="Result rank")
+    score: float = Field(..., description="Similarity score to query")
+    # rank is not strictly needed if we return a list, but keeping it if UI uses it
+    # rank: int = Field(..., description="Result rank") # Removing rank as it's implied by order
 
 class SlackFetchRequest(BaseModel):
     """Request model for fetching Slack messages"""

supabase_job_storage.py

@@ -0,0 +1,332 @@
+"""
+Supabase-based job storage for clustering jobs.
+Prevents memory leaks by storing jobs in PostgreSQL instead of in-memory dictionaries.
+"""
+import json
+import logging
+from datetime import datetime, timedelta
+from typing import Optional, Dict, Any, List
+from uuid import UUID
+
+from tenacity import retry, stop_after_attempt, wait_exponential
+from pydantic import BaseModel
+
+from database import get_client
+
+logger = logging.getLogger(__name__)
+
+
+class SupabaseJobStorage:
+    """
+    Manages clustering job storage in Supabase (PostgreSQL).
+    
+    Features:
+    - Persistent storage (survives server restarts)
+    - Automatic cleanup of old jobs
+    - Supports multiple servers
+    - Transaction safety
+    """
+    
+    def __init__(self, retention_hours: int = 48):
+        """
+        Initialize Supabase job storage.
+        
+        Args:
+            retention_hours: How long to keep jobs before cleanup (default 48 hours)
+        """
+        self.client = get_client()
+        self.retention_hours = retention_hours
+        
+        if self.client is None:
+            logger.warning("Supabase client not configured. Job storage will fail.")
+    
+    @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
+    def create_job(
+        self,
+        job_id: str,
+        status: str = "processing",
+        progress: float = 0.0,
+        message: str = "",
+        user_id: Optional[str] = None,
+        distance_threshold: Optional[float] = None,
+        min_cluster_size: Optional[int] = None,
+        force_recluster: bool = False
+    ) -> bool:
+        """Create a new clustering job in the database."""
+        if self.client is None:
+            return False
+        
+        try:
+            payload = {
+                "job_id": job_id,
+                "status": status,
+                "progress": progress,
+                "message": message,
+                "user_id": user_id,
+                "distance_threshold": distance_threshold,
+                "min_cluster_size": min_cluster_size,
+                "force_recluster": force_recluster
+            }
+            
+            response = self.client.table("clustering_jobs").insert(payload).execute()
+            logger.info(f"Created job {job_id} in database")
+            return True
+        except Exception as e:
+            logger.error(f"Failed to create job {job_id}: {e}")
+            return False
+    
+    @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
+    def update_job_status(
+        self,
+        job_id: str,
+        status: Optional[str] = None,
+        progress: Optional[float] = None,
+        message: Optional[str] = None
+    ) -> bool:
+        """Update job status, progress, or message."""
+        if self.client is None:
+            return False
+        
+        try:
+            updates: Dict[str, Any] = {}
+            if status is not None:
+                updates["status"] = status
+                if status == "completed":
+                    updates["completed_at"] = datetime.utcnow().isoformat()
+            if progress is not None:
+                updates["progress"] = progress
+            if message is not None:
+                updates["message"] = message
+            
+            if not updates:
+                return True
+            
+            response = self.client.table("clustering_jobs")\
+                .update(updates)\
+                .eq("job_id", job_id)\
+                .execute()
+            
+            return True
+        except Exception as e:
+            logger.error(f"Failed to update job {job_id}: {e}")
+            return False
+    
+    @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
+    def get_job(self, job_id: str) -> Optional[Dict[str, Any]]:
+        """Get job by ID."""
+        if self.client is None:
+            return None
+        
+        try:
+            response = self.client.table("clustering_jobs")\
+                .select("*")\
+                .eq("job_id", job_id)\
+                .execute()
+            
+            if response.data and len(response.data) > 0:
+                return response.data[0]
+            return None
+        except Exception as e:
+            logger.error(f"Failed to get job {job_id}: {e}")
+            return None
+    
+    @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
+    def save_result(self, job_id: str, result_data: Dict[str, Any]) -> bool:
+        """Save clustering result for a job."""
+        if self.client is None:
+            return False
+        
+        try:
+            payload = {
+                "job_id": job_id,
+                "result_data": result_data  # Supabase automatically handles JSONB
+            }
+            
+            # Use upsert to handle both insert and update cases
+            response = self.client.table("clustering_results")\
+                .upsert(payload)\
+                .execute()
+            
+            logger.info(f"Saved result for job {job_id}")
+            return True
+        except Exception as e:
+            logger.error(f"Failed to save result for job {job_id}: {e}")
+            return False
+    
+    @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
+    def get_result(self, job_id: str) -> Optional[Dict[str, Any]]:
+        """Get clustering result by job ID."""
+        if self.client is None:
+            return None
+        
+        try:
+            response = self.client.table("clustering_results")\
+                .select("result_data")\
+                .eq("job_id", job_id)\
+                .execute()
+            
+            if response.data and len(response.data) > 0:
+                return response.data[0]["result_data"]
+            return None
+        except Exception as e:
+            logger.error(f"Failed to get result for job {job_id}: {e}")
+            return None
+    
+    @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
+    def cleanup_old_jobs(self, hours: Optional[int] = None) -> int:
+        """
+        Delete jobs older than specified hours.
+        
+        Args:
+            hours: Retention period in hours (uses instance default if None)
+            
+        Returns:
+            Number of jobs deleted
+        """
+        if self.client is None:
+            return 0
+        
+        hours = hours or self.retention_hours
+        
+        try:
+            # Call the PostgreSQL function
+            response = self.client.rpc("cleanup_old_clustering_jobs", {
+                "retention_hours": hours
+            }).execute()
+            
+            deleted_count = response.data if response.data is not None else 0
+            
+            if deleted_count > 0:
+                logger.info(f"Cleaned up {deleted_count} old clustering jobs")
+            
+            return deleted_count
+        except Exception as e:
+            # Fallback: manual deletion if function doesn't exist
+            logger.warning(f"RPC function not available, using manual cleanup: {e}")
+            
+            try:
+                cutoff = (datetime.utcnow() - timedelta(hours=hours)).isoformat()
+                
+                # Get jobs to delete
+                old_jobs = self.client.table("clustering_jobs")\
+                    .select("job_id")\
+                    .lt("created_at", cutoff)\
+                    .execute()
+                
+                job_ids = [job["job_id"] for job in (old_jobs.data or [])]
+                
+                if job_ids:
+                    # Delete results first (due to foreign key)
+                    self.client.table("clustering_results")\
+                        .delete()\
+                        .in_("job_id", job_ids)\
+                        .execute()
+                    
+                    # Delete jobs
+                    self.client.table("clustering_jobs")\
+                        .delete()\
+                        .in_("job_id", job_ids)\
+                        .execute()
+                    
+                    logger.info(f"Manually cleaned up {len(job_ids)} old clustering jobs")
+                    return len(job_ids)
+                
+                return 0
+            except Exception as e2:
+                logger.error(f"Manual cleanup also failed: {e2}")
+                return 0
+    
+    @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
+    def get_recent_jobs(self, limit: int = 10, user_id: Optional[str] = None) -> List[Dict[str, Any]]:
+        """
+        Get recent jobs, optionally filtered by user.
+        
+        Args:
+            limit: Maximum number of jobs to return
+            user_id: Optional user ID to filter by
+            
+        Returns:
+            List of job dictionaries
+        """
+        if self.client is None:
+            return []
+        
+        try:
+            query = self.client.table("clustering_jobs")\
+                .select("*")\
+                .order("created_at", desc=True)\
+                .limit(limit)
+            
+            if user_id:
+                query = query.eq("user_id", user_id)
+            
+            response = query.execute()
+            return response.data or []
+        except Exception as e:
+            logger.error(f"Failed to get recent jobs: {e}")
+            return []
+    
+    def get_stats(self) -> Dict[str, Any]:
+        """Get storage statistics."""
+        if self.client is None:
+            return {
+                "total_jobs": 0,
+                "active_jobs": 0,
+                "completed_jobs": 0,
+                "error_jobs": 0,
+                "storage_type": "supabase (not connected)"
+            }
+        
+        try:
+            # Count jobs by status
+            all_jobs = self.client.table("clustering_jobs")\
+                .select("status", count="exact")\
+                .execute()
+            
+            total = all_jobs.count if hasattr(all_jobs, 'count') else len(all_jobs.data or [])
+            
+            active = self.client.table("clustering_jobs")\
+                .select("job_id", count="exact")\
+                .eq("status", "processing")\
+                .execute()
+            
+            completed = self.client.table("clustering_jobs")\
+                .select("job_id", count="exact")\
+                .eq("status", "completed")\
+                .execute()
+            
+            errors = self.client.table("clustering_jobs")\
+                .select("job_id", count="exact")\
+                .eq("status", "error")\
+                .execute()
+            
+            return {
+                "total_jobs": total,
+                "active_jobs": len(active.data or []),
+                "completed_jobs": len(completed.data or []),
+                "error_jobs": len(errors.data or []),
+                "storage_type": "supabase (postgresql)",
+                "retention_hours": self.retention_hours
+            }
+        except Exception as e:
+            logger.error(f"Failed to get stats: {e}")
+            return {
+                "total_jobs": 0,
+                "active_jobs": 0,
+                "completed_jobs": 0,
+                "error_jobs": 0,
+                "storage_type": "supabase (error)",
+                "error": str(e)
+            }
+
+
+# Global instance
+_storage_instance: Optional[SupabaseJobStorage] = None
+
+
+def get_job_storage() -> SupabaseJobStorage:
+    """Get or create the global job storage instance."""
+    global _storage_instance
+    if _storage_instance is None:
+        _storage_instance = SupabaseJobStorage()
+    return _storage_instance