Real-time streaming: /run-async with BackgroundTasks, UUID-first, SSE-delivered results

FRRONTEEEND/components/ChatInterface.tsx

@@ -101,8 +101,13 @@ export const ChatInterface: React.FC<{ onBack: () => void }> = ({ onBack }) => {
           // 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
+          console.log('✅ Analysis completed', data.result);
+          setIsTyping(false);
+          
+          // Process the final result
+          if (data.result) {
+            processAnalysisResult(data.result);
+          }
         }
       } catch (err) {
         console.error('❌ Error parsing SSE event:', err, e.data);
@@ -128,6 +133,74 @@ export const ChatInterface: React.FC<{ onBack: () => void }> = ({ onBack }) => {
     };
   }, [activeSessionId]);
 
+  const processAnalysisResult = (result: any) => {
+    // Extract and display the analysis result from SSE
+    let assistantContent = '✅ Analysis Complete!\n\n';
+    let reports: Array<{name: string, path: string}> = [];
+    let plots: Array<{title: string, url: string, type?: 'image' | 'html'}> = [];
+    
+    // Extract plots and reports from workflow_history
+    if (result.workflow_history) {
+      const reportTools = ['generate_ydata_profiling_report', 'generate_plotly_dashboard', 'generate_all_plots'];
+      const plotTools = [
+        'generate_interactive_correlation_heatmap',
+        'generate_interactive_scatter',
+        'generate_interactive_histogram',
+        'generate_interactive_box_plots',
+        'generate_interactive_time_series',
+        'generate_eda_plots',
+        'generate_data_quality_plots',
+        'analyze_correlations'
+      ];
+      
+      result.workflow_history.forEach((step: any) => {
+        if (reportTools.includes(step.tool)) {
+          const reportPath = step.result?.output_path || step.result?.report_path || step.arguments?.output_path;
+          if (reportPath && (step.result?.success !== false)) {
+            reports.push({
+              name: step.tool.replace('generate_', '').replace(/_/g, ' ').trim(),
+              path: reportPath
+            });
+          }
+        }
+        
+        if (plotTools.includes(step.tool) && step.result?.plots) {
+          step.result.plots.forEach((plot: any) => {
+            plots.push({
+              title: plot.title || plot.type || 'Plot',
+              url: plot.url || plot.path,
+              type: plot.url?.endsWith('.html') ? 'html' : 'image'
+            });
+          });
+        }
+      });
+    }
+    
+    if (reports.length > 0) {
+      assistantContent += '📊 **Generated Reports:**\n';
+      reports.forEach(r => assistantContent += `- ${r.name}\n`);
+      assistantContent += '\n';
+    }
+    
+    if (plots.length > 0) {
+      assistantContent += `📈 **Generated ${plots.length} Visualizations**\n\n`;
+    }
+    
+    assistantContent += result.final_answer || 'Analysis complete. Check the generated artifacts.';
+    
+    // Add assistant message with result
+    const assistantMessage: Message = {
+      id: Date.now().toString(),
+      role: 'assistant',
+      content: assistantContent,
+      timestamp: new Date(),
+      reports,
+      plots
+    };
+    
+    updateSession(activeSessionId, [...activeSession.messages, assistantMessage]);
+  };
+
   const handleSend = async () => {
     if ((!input.trim() && !uploadedFile) || isTyping) return;
 
@@ -183,7 +256,7 @@ export const ChatInterface: React.FC<{ onBack: () => void }> = ({ onBack }) => {
         formData.append('use_cache', 'true');
         formData.append('max_iterations', '20');
         
-        response = await fetch(`${API_URL}/run`, {
+        response = await fetch(`${API_URL}/run-async`, {
           method: 'POST',
           body: formData
         });
@@ -230,6 +303,14 @@ export const ChatInterface: React.FC<{ onBack: () => void }> = ({ onBack }) => {
         setActiveSessionId(data.session_id);
       }
       
+      // For async endpoint, result comes via SSE analysis_complete event
+      // For now, just wait for SSE to deliver the result
+      if (data.status === 'started') {
+        console.log('🚀 Analysis started, waiting for SSE events...');
+        return; // Don't process result here, will come via SSE
+      }
+      
+      // Legacy sync endpoint handling (if data.result exists)
       let assistantContent = '';
       let reports: Array<{name: string, path: string}> = [];
       let plots: Array<{title: string, url: string, type?: 'image' | 'html'}> = [];

src/api/app.py

@@ -16,7 +16,7 @@ from dotenv import load_dotenv
 # Load environment variables from .env file
 load_dotenv()
 
-from fastapi import FastAPI, File, UploadFile, Form, HTTPException, Request
+from fastapi import FastAPI, File, UploadFile, Form, HTTPException, Request, BackgroundTasks
 from fastapi.responses import JSONResponse, FileResponse, StreamingResponse
 from fastapi.staticfiles import StaticFiles
 from fastapi.middleware.cors import CORSMiddleware
@@ -278,6 +278,90 @@ class AnalysisRequest(BaseModel):
     max_iterations: int = 20
 
 
+def run_analysis_background(file_path: str, task_description: str, target_col: Optional[str], 
+                            use_cache: bool, max_iterations: int, session_id: str):
+    """Background task to run analysis and emit events."""
+    try:
+        logger.info(f"[BACKGROUND] Starting analysis for session {session_id}")
+        
+        result = agent.analyze(
+            file_path=file_path,
+            task_description=task_description,
+            target_col=target_col,
+            use_cache=use_cache,
+            max_iterations=max_iterations
+        )
+        
+        logger.info(f"[BACKGROUND] Analysis completed for session {session_id}")
+        
+        # Send completion event
+        progress_manager.emit(session_id, {
+            "type": "analysis_complete",
+            "status": result.get("status"),
+            "message": "✅ Analysis completed successfully!",
+            "result": result
+        })
+        
+    except Exception as e:
+        logger.error(f"[BACKGROUND] Analysis failed for session {session_id}: {e}")
+        progress_manager.emit(session_id, {
+            "type": "analysis_failed",
+            "error": str(e),
+            "message": f"❌ Analysis failed: {str(e)}"
+        })
+
+
+@app.post("/run-async")
+async def run_analysis_async(
+    background_tasks: BackgroundTasks,
+    file: Optional[UploadFile] = File(None),
+    task_description: str = Form(...),
+    target_col: Optional[str] = Form(None),
+    use_cache: bool = Form(True),
+    max_iterations: int = Form(20)
+) -> JSONResponse:
+    """
+    Start analysis in background and return session UUID immediately.
+    Frontend can connect SSE with this UUID to receive real-time updates.
+    """
+    if agent is None:
+        raise HTTPException(status_code=503, detail="Agent not initialized")
+    
+    # Get session UUID immediately
+    session_id = agent.session.session_id if hasattr(agent, 'session') and agent.session else "default"
+    logger.info(f"[ASYNC] Created session: {session_id}")
+    
+    # Handle file upload
+    temp_file_path = None
+    if file:
+        temp_dir = Path("/tmp") / "data_science_agent"
+        temp_dir.mkdir(parents=True, exist_ok=True)
+        temp_file_path = temp_dir / file.filename
+        
+        with open(temp_file_path, "wb") as buffer:
+            shutil.copyfileobj(file.file, buffer)
+        
+        logger.info(f"[ASYNC] File saved: {file.filename}")
+    
+    # Start background analysis
+    background_tasks.add_task(
+        run_analysis_background,
+        file_path=str(temp_file_path) if temp_file_path else "",
+        task_description=task_description,
+        target_col=target_col,
+        use_cache=use_cache,
+        max_iterations=max_iterations,
+        session_id=session_id
+    )
+    
+    # Return UUID immediately so frontend can connect SSE
+    return JSONResponse(content={
+        "session_id": session_id,
+        "status": "started",
+        "message": "Analysis started in background"
+    })
+
+
 @app.post("/run")
 async def run_analysis(
     file: Optional[UploadFile] = File(None, description="Dataset file (CSV or Parquet) - optional for follow-up requests"),