CRITICAL FIX: Implement real SSE event streaming with ProgressManager

- Created src/progress_manager.py with global event broadcaster
- Updated progress_callback to emit events via progress_manager
- Fixed SSE endpoint to stream events from progress_manager.subscribe()
- Updated frontend to handle onmessage events with proper debugging
- Added comprehensive logging: connection, received events, cleanup
- Events now properly flow: backend print -> emit -> SSE stream -> frontend

FRRONTEEEND/components/ChatInterface.tsx

@@ -66,6 +66,7 @@ export const ChatInterface: React.FC<{ onBack: () => void }> = ({ onBack }) => {
     if (!isTyping) {
       // Close SSE connection when workflow completes
       if (eventSourceRef.current) {
+        console.log('🔌 Closing SSE connection');
         eventSourceRef.current.close();
         eventSourceRef.current = null;
       }
@@ -82,50 +83,36 @@ export const ChatInterface: React.FC<{ onBack: () => void }> = ({ onBack }) => {
       console.log('✅ SSE connection established');
     };
 
-    // Handle connection event
-    eventSource.addEventListener('connected', (e) => {
-      console.log('Connected to progress stream:', e.data);
-    });
-
-    // Handle tool start events
-    eventSource.addEventListener('tool_start', (e) => {
-      try {
-        const data = JSON.parse(e.data);
-        setCurrentStep(data.message || `🔧 Executing: ${data.tool}`);
-      } catch (err) {
-        console.error('Error parsing tool_start event:', err);
-      }
-    });
-
-    // Handle tool complete events
-    eventSource.addEventListener('tool_complete', (e) => {
+    // Handle all incoming messages
+    eventSource.onmessage = (e) => {
+      console.log('📨 SSE received:', e.data);
       try {
         const data = JSON.parse(e.data);
-        setCurrentStep(data.message || `✓ Completed: ${data.tool}`);
-      } catch (err) {
-        console.error('Error parsing tool_complete event:', err);
-      }
-    });
-
-    // Handle tool error events
-    eventSource.addEventListener('tool_error', (e) => {
-      try {
-        const data = JSON.parse(e.data);
-        setCurrentStep(data.message || `❌ Failed: ${data.tool}`);
+        
+        // Handle different event types
+        if (data.type === 'connected') {
+          console.log('🔗 Connected to progress stream');
+        } else if (data.type === 'tool_executing') {
+          setCurrentStep(data.message || `🔧 Executing: ${data.tool}`);
+        } else if (data.type === 'tool_completed') {
+          setCurrentStep(data.message || `✓ Completed: ${data.tool}`);
+        } else if (data.type === 'tool_failed') {
+          setCurrentStep(data.message || `❌ Failed: ${data.tool}`);
+        } else if (data.type === 'token_update') {
+          // Optional: Display token budget updates
+          console.log('💰 Token update:', data.message);
+        } else if (data.type === 'analysis_complete') {
+          console.log('✅ Analysis completed');
+          setIsTyping(false);  // This will trigger cleanup
+        }
       } catch (err) {
-        console.error('Error parsing tool_error event:', err);
+        console.error('❌ Error parsing SSE event:', err, e.data);
       }
-    });
-
-    // Handle analysis completion
-    eventSource.addEventListener('analysis_complete', (e) => {
-      console.log('✅ Analysis completed');
-      setIsTyping(false);  // This will trigger cleanup
-    });
+    };
 
     // Handle errors
     eventSource.onerror = (err) => {
-      console.error('SSE error:', err);
+      console.error('❌ SSE error:', err);
       eventSource.close();
       eventSourceRef.current = null;
     };
@@ -135,6 +122,7 @@ export const ChatInterface: React.FC<{ onBack: () => void }> = ({ onBack }) => {
     // Cleanup on unmount or when isTyping changes to false
     return () => {
       if (eventSourceRef.current) {
+        console.log('🧹 Cleaning up SSE connection');
         eventSourceRef.current.close();
         eventSourceRef.current = null;
       }

src/api/app.py

@@ -26,6 +26,7 @@ import json
 
 # Import from parent package
 from src.orchestrator import DataScienceCopilot
+from src.progress_manager import progress_manager
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)
@@ -181,51 +182,46 @@ async def get_progress(session_id: str):
 async def stream_progress(session_id: str):
     """Stream real-time progress updates using Server-Sent Events (SSE).
     
-    This replaces the polling mechanism with a persistent connection that
-    receives events as they happen during workflow execution.
+    This endpoint connects clients to the global progress_manager which
+    receives events from the orchestrator as tools execute.
     
     Events:
-        - tool_start: When a tool begins execution
-        - tool_complete: When a tool finishes successfully
-        - tool_error: When a tool fails
+        - tool_executing: When a tool begins execution
+        - tool_completed: When a tool finishes successfully  
+        - tool_failed: When a tool fails
+        - token_update: Token budget updates
         - analysis_complete: When the entire workflow finishes
-        - status_update: General status messages
     """
     async def event_generator():
-        queue = event_manager.create_stream(session_id)
-        
         try:
             # Send initial connection event
-            yield f"event: connected\ndata: {{\"session_id\": \"{session_id}\"}}\n\n"
+            connection_event = {
+                'type': 'connected',
+                'message': '🔗 Connected to progress stream',
+                'session_id': session_id
+            }
+            yield f"data: {json.dumps(connection_event)}\n\n"
             
-            # Send current status if exists
-            current = event_manager.get_current_status(session_id)
-            if current:
-                yield f"event: {current['type']}\ndata: {json.dumps(current['data'])}\n\n"
+            # Send any existing history first (for reconnections)
+            history = progress_manager.get_history(session_id)
+            for event in history[-10:]:  # Send last 10 events
+                yield f"data: {json.dumps(event)}\n\n"
             
-            # Stream events as they arrive
-            while True:
-                try:
-                    # Wait for next event with timeout
-                    event_type, data = await asyncio.wait_for(queue.get(), timeout=30.0)
-                    
-                    # Check for completion signal
-                    if event_type == "complete":
-                        yield f"event: analysis_complete\ndata: {{\"status\": \"completed\"}}\n\n"
-                        break
-                    
-                    # Send the event
-                    yield f"event: {event_type}\ndata: {json.dumps(data)}\n\n"
-                    
-                except asyncio.TimeoutError:
-                    # Send keepalive ping
-                    yield f": keepalive\n\n"
-                    continue
+            # Stream new events as they occur
+            async for event in progress_manager.subscribe(session_id):
+                yield f"data: {json.dumps(event)}\n\n"
+                
+                # Check if analysis is complete
+                if event.get('type') == 'analysis_complete':
+                    break
                     
         except asyncio.CancelledError:
             logger.info(f"SSE stream cancelled for session {session_id}")
+            progress_manager.clear(session_id)
+        except Exception as e:
+            logger.error(f"SSE error for session {session_id}: {e}")
         finally:
-            event_manager.remove_stream(session_id, queue)
+            logger.info(f"SSE stream closed for session {session_id}")
     
     return StreamingResponse(
         event_generator(),
@@ -316,23 +312,17 @@ async def run_analysis(
                 "timestamp": time.time()
             })
             
-            # Send SSE event asynchronously
-            event_type = "tool_start" if status == "running" else "tool_complete" if status == "completed" else "tool_error"
+            # Emit event to progress_manager (synchronous, works in any context)
+            event_type = "tool_executing" if status == "running" else "tool_completed" if status == "completed" else "tool_failed"
             event_data = {
+                "type": event_type,
                 "tool": tool_name,
                 "status": status,
                 "message": f"🔧 Executing: {tool_name.replace('_', ' ').title()}" if status == "running" else 
                            f"✓ Completed: {tool_name.replace('_', ' ').title()}" if status == "completed" else
-                           f"❌ Failed: {tool_name.replace('_', ' ').title()}",
-                "timestamp": time.time()
+                           f"❌ Failed: {tool_name.replace('_', ' ').title()}"
             }
-            
-            # Schedule the async event send
-            try:
-                asyncio.create_task(event_manager.send_event(session_key, event_type, event_data))
-            except RuntimeError:
-                # If no event loop, we're in sync context - that's ok, legacy polling still works
-                pass
+            progress_manager.emit(session_key, event_data)
         
         # Set progress callback on existing agent
         agent.progress_callback = progress_callback
@@ -350,18 +340,11 @@ async def run_analysis(
             logger.info(f"Follow-up analysis completed: {result.get('status')}")
             
             # Send completion event via SSE
-            try:
-                asyncio.create_task(event_manager.send_event(
-                    session_key, 
-                    "analysis_complete",
-                    {
-                        "status": result.get("status"),
-                        "message": "✅ Analysis completed successfully!",
-                        "timestamp": time.time()
-                    }
-                ))
-            except RuntimeError:
-                pass
+            progress_manager.emit(session_key, {
+                "type": "analysis_complete",
+                "status": result.get("status"),
+                "message": "✅ Analysis completed successfully!"
+            })
             
             # Make result JSON serializable
             def make_json_serializable(obj):
@@ -444,23 +427,17 @@ async def run_analysis(
                 "timestamp": time.time()
             })
             
-            # Send SSE event asynchronously
-            event_type = "tool_start" if status == "running" else "tool_complete" if status == "completed" else "tool_error"
+            # Emit event to progress_manager (synchronous, works in any context)
+            event_type = "tool_executing" if status == "running" else "tool_completed" if status == "completed" else "tool_failed"
             event_data = {
+                "type": event_type,
                 "tool": tool_name,
                 "status": status,
                 "message": f"🔧 Executing: {tool_name.replace('_', ' ').title()}" if status == "running" else 
                            f"✓ Completed: {tool_name.replace('_', ' ').title()}" if status == "completed" else
-                           f"❌ Failed: {tool_name.replace('_', ' ').title()}",
-                "timestamp": time.time()
+                           f"❌ Failed: {tool_name.replace('_', ' ').title()}"
             }
-            
-            # Schedule the async event send
-            try:
-                asyncio.create_task(event_manager.send_event(session_key, event_type, event_data))
-            except RuntimeError:
-                # If no event loop, we're in sync context - that's ok, legacy polling still works
-                pass
+            progress_manager.emit(session_key, event_data)
         
         # Set progress callback on existing agent
         agent.progress_callback = progress_callback
@@ -478,18 +455,11 @@ async def run_analysis(
         logger.info(f"Analysis completed: {result.get('status')}")
         
         # Send completion event via SSE
-        try:
-            asyncio.create_task(event_manager.send_event(
-                session_key, 
-                "analysis_complete",
-                {
-                    "status": result.get("status"),
-                    "message": "✅ Analysis completed successfully!",
-                    "timestamp": time.time()
-                }
-            ))
-        except RuntimeError:
-            pass
+        progress_manager.emit(session_key, {
+            "type": "analysis_complete",
+            "status": result.get("status"),
+            "message": "✅ Analysis completed successfully!"
+        })
         
         # Filter out non-JSON-serializable objects (like matplotlib/plotly Figures)
         def make_json_serializable(obj):

src/progress_manager.py

@@ -0,0 +1,131 @@
+"""
+Global Progress Event Manager for Real-Time SSE Streaming
+
+This module provides a singleton ProgressManager that captures all workflow progress
+events and broadcasts them to connected SSE clients in real-time.
+"""
+
+import asyncio
+import json
+from typing import Dict, List, Any, Optional
+from datetime import datetime
+from collections import defaultdict
+
+
+class ProgressManager:
+    """
+    Manages progress events for active analysis sessions.
+    
+    Features:
+    - Emit events to multiple subscribers simultaneously
+    - Store event history for late-joining clients
+    - Automatic cleanup of dead connections
+    - Thread-safe event broadcasting
+    """
+    
+    def __init__(self):
+        self._queues: Dict[str, List[asyncio.Queue]] = defaultdict(list)
+        self._history: Dict[str, List[Dict]] = defaultdict(list)
+        self._lock = asyncio.Lock()
+    
+    def emit(self, session_id: str, event: Dict[str, Any]):
+        """
+        Emit a progress event to all subscribers.
+        
+        Args:
+            session_id: Session identifier
+            event: Event data (must include 'type' and 'message')
+        """
+        # Add timestamp
+        event['timestamp'] = datetime.now().isoformat()
+        
+        # Store in history
+        self._history[session_id].append(event)
+        
+        # Limit history size to prevent memory leaks
+        if len(self._history[session_id]) > 500:
+            self._history[session_id] = self._history[session_id][-500:]
+        
+        # Send to all active subscribers
+        if session_id in self._queues:
+            dead_queues = []
+            for queue in self._queues[session_id]:
+                try:
+                    queue.put_nowait(event)
+                except asyncio.QueueFull:
+                    # Queue is full, mark for removal
+                    dead_queues.append(queue)
+                except Exception:
+                    dead_queues.append(queue)
+            
+            # Remove dead queues
+            for dead_queue in dead_queues:
+                if dead_queue in self._queues[session_id]:
+                    self._queues[session_id].remove(dead_queue)
+    
+    async def subscribe(self, session_id: str):
+        """
+        Subscribe to progress events for a session.
+        
+        Args:
+            session_id: Session identifier
+            
+        Yields:
+            Progress events as they occur
+        """
+        queue = asyncio.Queue(maxsize=100)
+        self._queues[session_id].append(queue)
+        
+        try:
+            while True:
+                event = await queue.get()
+                yield event
+        except asyncio.CancelledError:
+            # Client disconnected
+            pass
+        finally:
+            # Cleanup
+            if session_id in self._queues and queue in self._queues[session_id]:
+                self._queues[session_id].remove(queue)
+    
+    def get_history(self, session_id: str) -> List[Dict]:
+        """
+        Get all past events for a session.
+        
+        Args:
+            session_id: Session identifier
+            
+        Returns:
+            List of past events
+        """
+        return self._history.get(session_id, [])
+    
+    def clear(self, session_id: str):
+        """
+        Clear history and disconnect all subscribers for a session.
+        
+        Args:
+            session_id: Session identifier
+        """
+        if session_id in self._history:
+            del self._history[session_id]
+        if session_id in self._queues:
+            # Close all queues
+            for queue in self._queues[session_id]:
+                try:
+                    queue.put_nowait({'type': 'session_cleared', 'message': 'Session ended'})
+                except:
+                    pass
+            del self._queues[session_id]
+    
+    def get_active_sessions(self) -> List[str]:
+        """Get list of sessions with active subscribers."""
+        return [sid for sid, queues in self._queues.items() if len(queues) > 0]
+    
+    def get_subscriber_count(self, session_id: str) -> int:
+        """Get number of active subscribers for a session."""
+        return len(self._queues.get(session_id, []))
+
+
+# Global singleton instance
+progress_manager = ProgressManager()