feat: Implement remaining V3 architecture fixes

Fixes 2, 3, and 5 implemented in app.py:

Pipeline Record Creation (Fix 2):
- Import pipeline_manager and workflow_manager
- Create pipeline record BEFORE execution starts
- Use execution_id instead of session_id for tracking

S3-Only Result Storage (Fix 3):
- Removed pipeline_result storage in session documents
- Removed deprecated save_pipeline_execution() calls
- Results now stored only in S3 via pipeline_manager
- MongoDB pipelines collection properly populated

Workflow Save Prompting (Fix 5):
- Added prompt after successful execution
- Stores pending_workflow_save in session
- Detects user confirmation (save/yes/sure/ok)
- Saves workflow to workflows collection + S3
- Clears pending state after save

Component Status Handling:
- Added component_status event type handling
- Displays component success messages to user
- User-friendly emoji status indicators

Chat Name Fix:
- Only generates once per session (already committed)

Remaining:
- Fix 4: Gradio file upload integration (separate commit)

app.py

@@ -670,6 +670,10 @@ from services.intent_classifier import intent_classifier
 from api_routes import router as api_router
 from api_routes_v2 import router as api_router_v2
 
+# V3 Architecture managers
+from services.pipeline_manager import get_pipeline_manager
+from services.workflow_manager import get_workflow_manager
+
 
 # ========================
 # BACKGROUND CLEANUP TASK
@@ -859,6 +863,54 @@ def chatbot_response_streaming(message: str, history: List, session_id: str, fil
         yield format_chat_history(history, message, friendly_response)
         return
     
+    # ========================
+    # HANDLE WORKFLOW SAVE REQUEST (V3)
+    # ========================
+    pending_workflow = session.get("pending_workflow_save")
+    if pending_workflow and any(phrase in message.lower() for phrase in ["save", "yes", "sure", "ok"]):
+        try:
+            from services.s3_manager import get_s3_manager
+            
+            pipeline_mgr = get_pipeline_manager()
+            workflow_mgr = get_workflow_manager()
+            s3 = get_s3_manager()
+            
+            # Get pipeline from pipelines collection
+            pipeline_record = pipeline_mgr.get_pipeline(pending_workflow["execution_id"])
+            
+            if pipeline_record and pipeline_record.get("pipeline_definition_s3_key"):
+                # Download pipeline definition from S3
+                pipeline_def = s3.download_json(pipeline_record["pipeline_definition_s3_key"], add_prefix=False)
+                
+                # Save as workflow
+                workflow_id = workflow_mgr.save_workflow(
+                    session_id=session_id,
+                    pipeline_definition=pipeline_def,
+                    user_message=message
+                )
+                
+                # Clear pending
+                session_manager.update_session(session_id, {"pending_workflow_save": None})
+                
+                response = f"✅ **Workflow Saved!**\n\nWorkflow ID: `{workflow_id}`\nName: {pending_workflow['pipeline_name']}\n\nYou can now reuse this workflow anytime!\n\nWhat else can I help you with?"
+                session_manager.add_message(session_id, "assistant", response)
+                yield format_chat_history(history, message, response)
+                return
+            else:
+                # Pipeline record not found
+                session_manager.update_session(session_id, {"pending_workflow_save": None})
+                response = "⚠️ Sorry, I couldn't find the pipeline to save. The workflow save request has expired.\n\nWhat else can I help you with?"
+                session_manager.add_message(session_id, "assistant", response)
+                yield format_chat_history(history, message, response)
+                return
+                
+        except Exception as e:
+            session_manager.update_session(session_id, {"pending_workflow_save": None})
+            response = f"❌ Failed to save workflow: {str(e)}\n\nWhat would you like to do next?"
+            session_manager.add_message(session_id, "assistant", response)
+            yield format_chat_history(history, message, response)
+            return
+    
     # ========================
     # STATE: INITIAL - Generate Pipeline ONLY if intent requires it
     # ========================
@@ -961,7 +1013,35 @@ Here's what I'll do:
                         accumulated_response = initial_message + "\n\n" + "\n".join(progress_messages)
                         yield format_chat_history(history, message, accumulated_response)
                     
-                    # Step updates
+                    # Component status updates (V3)
+                    elif event_type == "component_status":
+                        step_info = {
+                            "step": event.get("step", 0),
+                            "component": event.get("component", "processing"),
+                            "status": event.get("status", "running"),
+                            "message": event.get("message", ""),
+                            "executor": event.get("executor", "unknown")
+                        }
+                        
+                        if "observation" in event:
+                            step_info["observation"] = event.get("observation")
+                        
+                        steps_completed.append(step_info)
+                        executor_used = event.get("executor", executor_used)
+                        
+                        # User-friendly status
+                        status_emoji = "✅" if event.get('status') == 'completed' else "⏳"
+                        component_msg = event.get('message', '')
+                        status_line = f"{status_emoji} Step {event.get('step', 0)}: {event.get('component', 'processing')}"
+                        if component_msg:
+                            status_line += f" - {component_msg}"
+                        
+                        progress_messages.append(status_line)
+                        current_progress = "\n".join(progress_messages[-5:])  # Last 5 steps
+                        accumulated = initial_message + "\n\n" + current_progress
+                        yield format_chat_history(history, message, accumulated)
+                    
+                    # Legacy step updates (backwards compatibility)
                     elif event_type == "step":
                         step_info = {
                             "step": event.get("step", 0),
@@ -1015,22 +1095,28 @@ Here's what I'll do:
                 
                 # Process final result
                 if final_payload:
+                    # V3: DO NOT store results in session - only in S3 via pipeline_manager
+                    # The pipeline_manager already handled S3 storage during execution
                     session_manager.update_session(session_id, {
-                        "pipeline_result": final_payload,
                         "state": ConversationState.INITIAL
                     })
                     
-                    # Save execution to MongoDB
-                    session_manager.save_pipeline_execution(
-                        session_id=session_id,
-                        pipeline=plan,
-                        result=final_payload,
-                        file_path=session.get("current_file"),
-                        executor=executor_used
-                    )
+                    # V3: Store pending workflow save info
+                    session_manager.update_session(session_id, {
+                        "pending_workflow_save": {
+                            "execution_id": execution_id,
+                            "pipeline_name": plan.get("pipeline_name", "Untitled")
+                        }
+                    })
+                    
+                    # REMOVED: Deprecated save_pipeline_execution - now handled by pipeline_manager
                     
-                    # Create user-friendly final response
+                    # Create user-friendly final response with workflow save prompt
                     success_count = len([s for s in steps_completed if s.get("status") == "completed"])
+                    
+                    # V3: Check if we have S3 final output URL
+                    final_output_url = final_payload.get("final_output_url", "")
+                    
                     friendly_final = f"""🎉 **Pipeline Completed Successfully!**
 
 **Summary:**
@@ -1039,7 +1125,10 @@ Here's what I'll do:
 - Successful: {success_count}
 - Executor: {executor_used}
 
-✅ All done! What else would you like me to help you with?
+💾 **Would you like to save this workflow for future use?**
+Type **'save workflow'** or **'yes'** to save it.
+
+✅ All done! What else can I help you with?
 
 <details>
 <summary>📊 Detailed Results (for developers)</summary>