adding pipeline status into the message response

api_routes_v2.py

@@ -645,13 +645,24 @@ def _add_and_mirror_message(
     content: str,
     *,
     pipeline_id: Optional[str] = None,
+    pipeline_action: Optional[str] = None,  # "created" | "executed" | "failed"
     pipeline_status: Optional[str] = None,
     pipeline_result: Optional[dict] = None,
     file_metadata: Optional[dict] = None,
 ):
     """
     V3 RULE: Append message to S3 conversation.
-    Now includes pipeline results in assistant messages.
+    Now properly includes pipeline_id and results in messages.
+    
+    Args:
+        chat_id: Session ID
+        role: "user" | "assistant" | "system"
+        content: Message text
+        pipeline_id: ID of related pipeline (if any)
+        pipeline_action: "created" | "executed" | "failed"
+        pipeline_status: Pipeline status (for executed pipelines)
+        pipeline_result: Full pipeline result (for executed pipelines)
+        file_metadata: File info if message has file attachment
     """
     # 1. Load existing
     current_messages = _load_conversation_from_s3(chat_id)
@@ -663,34 +674,44 @@ def _add_and_mirror_message(
         "timestamp": datetime.utcnow().isoformat() + "Z"
     }
     
-    # NEW: Add pipeline results if available (for assistant messages)
-    if role == "assistant" and pipeline_result:
-        # Check if this is a pipeline completion message
-        if "pipeline completed" in content.lower() or "pipeline failed" in content.lower():
-            # Extract the text/result from pipeline
-            execution_results = pipeline_result.get("execution_results", {})
-            
-            # Try to extract the main result text
+    # ✅ FIX: Add pipeline_id if provided
+    if pipeline_id:
+        new_msg["pipeline_id"] = pipeline_id
+    
+    # ✅ FIX: Add pipeline_action if provided
+    if pipeline_action:
+        new_msg["pipeline_action"] = pipeline_action
+    
+    # ✅ FIX: Add result for executed/failed pipelines
+    if role == "assistant" and pipeline_action in ("executed", "failed") and pipeline_id:
+        result_data = {
+            "pipeline_id": pipeline_id,
+            "status": pipeline_status or (pipeline_result.get("status") if pipeline_result else None) or pipeline_action,
+        }
+        
+        if pipeline_result:
+            # Extract user-facing text from result
             result_text = ""
+            execution_results = pipeline_result.get("execution_results", pipeline_result)
             
+            # Try to extract the main result text from various locations
             # Check for image descriptions
             if execution_results.get("image_descriptions"):
                 image_desc = execution_results["image_descriptions"]
                 if image_desc.get("result") and isinstance(image_desc["result"], list) and len(image_desc["result"]) > 0:
                     page_result = image_desc["result"][0]
-                    # Use Gemini description if available
                     if page_result.get("gemini", {}).get("description"):
                         result_text = page_result["gemini"]["description"]
                     elif page_result.get("mistral", {}).get("description"):
                         result_text = page_result["mistral"]["description"]
             
-            # Check for extracted text (for text extraction pipelines)
+            # Check for extracted text
             elif execution_results.get("text"):
                 result_text = execution_results["text"]
             
             # Check for component results
             elif execution_results.get("components_executed"):
-                for comp in execution_results["components_executed"]:
+                for comp in execution_results.get("components_executed", []):
                     comp_result = comp.get("result", {})
                     if comp_result.get("text"):
                         result_text = comp_result["text"]
@@ -706,20 +727,26 @@ def _add_and_mirror_message(
                                 result_text = page_result["mistral"]["description"]
                                 break
             
-            # Add result to message if we found something
-            if result_text:
-                new_msg["result"] = {
-                    "text": result_text,
-                    "pipeline_id": pipeline_result.get("pipeline_id"),
-                    "status": pipeline_result.get("status", "completed")
-                }
-            elif pipeline_result.get("error"):
-                # For error cases
-                new_msg["result"] = {
-                    "error": pipeline_result.get("error"),
-                    "pipeline_id": pipeline_result.get("pipeline_id"),
-                    "status": "failed"
-                }
+            # Fallback to generic extraction
+            if not result_text:
+                result_text = _extract_user_facing_text(pipeline_result)
+            
+            result_data["text"] = result_text
+            
+            # Add error if present
+            if pipeline_result.get("error"):
+                result_data["error"] = pipeline_result.get("error")
+            
+            # Add summary stats
+            if pipeline_result.get("completed_steps") is not None:
+                result_data["completed_steps"] = pipeline_result.get("completed_steps")
+            if pipeline_result.get("total_steps") is not None:
+                result_data["total_steps"] = pipeline_result.get("total_steps")
+        
+        elif pipeline_action == "failed":
+            result_data["error"] = "Pipeline execution failed"
+        
+        new_msg["result"] = result_data
     
     # Add file metadata if provided
     if file_metadata:
@@ -750,22 +777,45 @@ def _assistant_response_payload(
     final_output: Optional[Dict[str, Any]] = None,
     exception: Optional[str] = None,
     pipeline_result: Optional[Dict[str, Any]] = None,
-    pipeline_id: Optional[str] = None,  # ✅ ADD THIS PARAMETER
+    pipeline_id: Optional[str] = None,
+    pipeline_action: Optional[str] = None,  # ✅ NEW: "created" | "executed" | "failed"
 ) -> ChatResponse:
     """
     Create ChatResponse payload with all required fields.
+    
+    Args:
+        chat_id: Session ID
+        friendly_response: User-friendly message
+        intent: Intent classification result
+        api_data: Technical API response data
+        state: Current session state
+        output: Pipeline summary output
+        final_output: Final downloadable result
+        exception: Error message if any
+        pipeline_result: Full pipeline execution result
+        pipeline_id: Pipeline ID (for tracking in messages)
+        pipeline_action: "created" | "executed" | "failed"
     """
     # Generate message_id for assistant response
     from services.schemas import generate_message_id
     message_id = generate_message_id()
     
-    # Persist assistant message to S3 WITH pipeline results AND pipeline_id
+    # Determine pipeline status from result or exception
+    pipeline_status = None
+    if pipeline_result:
+        pipeline_status = pipeline_result.get("status")
+    elif exception:
+        pipeline_status = "failed"
+    
+    # Persist assistant message to S3 WITH pipeline data
     _add_and_mirror_message(
         chat_id=chat_id,
         role="assistant",
         content=friendly_response,
-        pipeline_result=pipeline_result,  # Pass pipeline results
-        pipeline_id=pipeline_id  # ✅ PASS THE PIPELINE_ID HERE
+        pipeline_id=pipeline_id,
+        pipeline_action=pipeline_action,
+        pipeline_status=pipeline_status,
+        pipeline_result=pipeline_result,
     )
     
     # Get file metadata from session
@@ -1023,10 +1073,11 @@ async def get_session_history(
 ):
     """
     Get conversation history for a session.
-    Correct behavior:
-    - Preserve pipeline_id on messages
-    - Enrich ONLY using stored pipeline_id
-    - No inference, no guessing, no context scanning
+    
+    V3 FIX: 
+    - pipeline_id and pipeline_action are now stored IN the messages
+    - result is embedded for executed pipelines
+    - Only enrich if result is missing (fallback)
     """
     try:
         # Load conversation history from S3
@@ -1037,58 +1088,66 @@ async def get_session_history(
         for msg in history:
             msg_copy = msg.copy()
 
-            # ✅ ALWAYS pass through pipeline_id if present
-            pipeline_id = msg.get("pipeline_id")
-            if pipeline_id:
-                msg_copy["pipeline_id"] = pipeline_id
+            # Ensure message_id exists
+            if "message_id" not in msg_copy:
+                from services.schemas import generate_message_id
+                msg_copy["message_id"] = generate_message_id()
 
-            role = msg_copy.get("role")
-            content = msg_copy.get("content", "")
+            # Normalize timestamp
+            if "timestamp" in msg_copy and isinstance(msg_copy["timestamp"], datetime):
+                msg_copy["timestamp"] = msg_copy["timestamp"].isoformat()
 
-            # ✅ Enrich ONLY pipeline completion messages WITH pipeline_id
+            # ✅ pipeline_id and pipeline_action are now IN the message - no guessing needed!
+            pipeline_id = msg_copy.get("pipeline_id")
+            pipeline_action = msg_copy.get("pipeline_action")
+            
+            # ✅ Enrich ONLY if result is missing for executed/failed pipelines
             if (
-                role == "assistant"
-                and "Pipeline completed" in content
-                and pipeline_id
+                pipeline_id 
+                and pipeline_action in ("executed", "failed") 
+                and not msg_copy.get("result")
             ):
+                # Fallback: Load result from S3 pipeline file
                 try:
                     s3_key = f"{S3_PREFIX}/pipelines/{pipeline_id}.json"
                     resp = s3.get_object(Bucket=S3_BUCKET, Key=s3_key)
-                    pipeline_data = json.loads(
-                        resp["Body"].read().decode("utf-8")
-                    )
+                    pipeline_data = json.loads(resp["Body"].read().decode("utf-8"))
 
-                    # Attach final output if present
                     if "result" in pipeline_data:
-                        msg_copy["final_output"] = pipeline_data["result"]
+                        # Extract user-facing text
+                        result_text = _extract_user_facing_text(pipeline_data["result"])
+                        msg_copy["result"] = {
+                            "pipeline_id": pipeline_id,
+                            "status": pipeline_data.get("status", pipeline_action),
+                            "text": result_text,
+                        }
+                        if pipeline_data["result"].get("error"):
+                            msg_copy["result"]["error"] = pipeline_data["result"]["error"]
+                except Exception as e:
+                    # Non-fatal enrichment failure
+                    msg_copy["result"] = {
+                        "pipeline_id": pipeline_id,
+                        "status": "unknown",
+                        "error": f"Could not load result: {str(e)}"
+                    }
 
-                    # Attach pipeline metadata
+            # ✅ Add pipeline_metadata for created/executed pipelines
+            if pipeline_id:
+                try:
+                    s3_key = f"{S3_PREFIX}/pipelines/{pipeline_id}.json"
+                    resp = s3.get_object(Bucket=S3_BUCKET, Key=s3_key)
+                    pipeline_data = json.loads(resp["Body"].read().decode("utf-8"))
+                    
                     msg_copy["pipeline_metadata"] = {
                         "pipeline_id": pipeline_id,
-                        "pipeline_name": pipeline_data.get(
-                            "pipeline_name", "unknown"
-                        ),
+                        "pipeline_name": pipeline_data.get("pipeline_name", "unknown"),
                         "status": pipeline_data.get("status", "unknown"),
-                        "executed_at": pipeline_data.get(
-                            "updated_at",
-                            pipeline_data.get("created_at")
-                        )
+                        "created_at": pipeline_data.get("created_at"),
+                        "updated_at": pipeline_data.get("updated_at"),
                     }
-
-                except Exception as e:
-                    # Non-fatal enrichment failure
-                    msg_copy["final_output_error"] = str(e)
-
-            # Ensure message_id exists
-            if "message_id" not in msg_copy:
-                from services.schemas import generate_message_id
-                msg_copy["message_id"] = generate_message_id()
-
-            # Normalize timestamp
-            if "timestamp" in msg_copy and isinstance(
-                msg_copy["timestamp"], datetime
-            ):
-                msg_copy["timestamp"] = msg_copy["timestamp"].isoformat()
+                except Exception:
+                    # Keep existing pipeline_id, just no extra metadata
+                    pass
 
             enhanced_history.append(msg_copy)
 
@@ -1099,7 +1158,7 @@ async def get_session_history(
         # Load session metadata
         session = session_manager.get_session(session_id) or {}
 
-        # ----- PIPELINES HISTORY (unchanged but cleaned) -----
+        # ----- PIPELINES HISTORY -----
         pipelines_hist = session.get("pipelines_history", [])
         enhanced_pipelines = []
 
@@ -1113,13 +1172,8 @@ async def get_session_history(
             pipeline_s3_key = pipeline_meta.get("pipeline_s3_key")
             if pipeline_s3_key:
                 try:
-                    resp = s3.get_object(
-                        Bucket=S3_BUCKET,
-                        Key=pipeline_s3_key
-                    )
-                    pipeline_def = json.loads(
-                        resp["Body"].read().decode("utf-8")
-                    )
+                    resp = s3.get_object(Bucket=S3_BUCKET, Key=pipeline_s3_key)
+                    pipeline_def = json.loads(resp["Body"].read().decode("utf-8"))
 
                     components = (
                         pipeline_def.get("components")
@@ -1141,16 +1195,13 @@ async def get_session_history(
                     enhanced_pipe["component_count"] = 0
                     enhanced_pipe["tools"] = []
                     enhanced_pipe["load_error"] = str(e)
-
             else:
                 enhanced_pipe["components"] = []
                 enhanced_pipe["component_count"] = 0
                 enhanced_pipe["tools"] = []
 
             if "hasError" not in enhanced_pipe:
-                enhanced_pipe["hasError"] = (
-                    enhanced_pipe.get("status") == "failed"
-                )
+                enhanced_pipe["hasError"] = enhanced_pipe.get("status") == "failed"
 
             enhanced_pipelines.append(enhanced_pipe)
 
@@ -1415,6 +1466,9 @@ async def chat_unified(
             file_ref = session.get("current_file")
             local_path, cleanup = download_to_temp_file(file_ref)
             session_manager.update_session(chat_id, {"state": "executing"})
+            
+            # ✅ Get pipeline_id from proposed pipeline
+            pipeline_id = proposed.get("pipeline_id")
 
             try:
                 result = execute_pipeline(
@@ -1435,9 +1489,7 @@ async def chat_unified(
                 is_success = (pipeline_status == "completed" and completed_steps == total_steps and not has_error)
                 
                 # V3: Update pipeline status in S3
-                pipeline_id = proposed.get("pipeline_id")
                 if pipeline_id:
-                    # Mark as failed if not successful, otherwise completed
                     final_status = "completed" if is_success else "failed"
                     _update_pipeline_status(pipeline_id, chat_id, final_status, result=result)
                     _record_model_attribution(
@@ -1496,7 +1548,7 @@ async def chat_unified(
                         "pipeline": api_pipeline
                     }
                     
-                    # Return response with pipeline result included
+                    # ✅ Return response with pipeline_id and pipeline_action
                     return _assistant_response_payload(
                         chat_id=chat_id,
                         friendly_response=friendly,
@@ -1506,8 +1558,9 @@ async def chat_unified(
                         output=output,
                         final_output=final_output,
                         exception=exception_msg,
-                        pipeline_result=result,  # NEW: Pass the pipeline results
-                        pipeline_id=pipeline_id  # ✅ PASS pipeline_id HERE
+                        pipeline_result=result,
+                        pipeline_id=pipeline_id,
+                        pipeline_action="executed"  # ✅ ADD THIS
                     )
                 else:
                     # Pipeline failed or partially completed
@@ -1523,7 +1576,6 @@ async def chat_unified(
                             })
                     
                     if failed_components:
-                        # Show specific component error
                         first_error = failed_components[0]
                         friendly = f"❌ Pipeline failed: {first_error['tool_name']} - {first_error['error']}"
                     else:
@@ -1533,20 +1585,20 @@ async def chat_unified(
                         "component_summary": f"Pipeline {pipeline_status}",
                         "steps": total_steps,
                         "completed": completed_steps,
-                        "failed": total_steps - completed_steps
+                        "failed": total_steps - completed_steps,
+                        "pipeline_id": pipeline_id
                     }
                     final_output = {"text": f"Pipeline execution {pipeline_status} with {completed_steps}/{total_steps} steps completed"}
                     api_type = "pipeline_failed" if pipeline_status == "failed" else "pipeline_partial"
                     exception_msg = error_msg
                     
-                    # Build api_response data for failure
                     api_data = {
                         "type": api_type,
                         "result": result,
                         "pipeline": proposed
                     }
                     
-                    # Return response with pipeline result included (even for failures)
+                    # ✅ Return response with pipeline_id and pipeline_action="failed"
                     return _assistant_response_payload(
                         chat_id=chat_id,
                         friendly_response=friendly,
@@ -1556,12 +1608,12 @@ async def chat_unified(
                         output=output,
                         final_output=final_output,
                         exception=exception_msg,
-                        pipeline_result=result,  # NEW: Pass results even for failures
-                        pipeline_id=pipeline_id  # ✅ PASS pipeline_id HERE
+                        pipeline_result=result,
+                        pipeline_id=pipeline_id,
+                        pipeline_action="failed"  # ✅ ADD THIS
                     )
             except Exception as e:
                 session_manager.update_session(chat_id, {"state": "initial"})
-                pipeline_id = proposed.get("pipeline_id")
                 if pipeline_id:
                     _update_pipeline_status(pipeline_id, chat_id, "failed", result={"error": str(e)})
                 
@@ -1573,7 +1625,6 @@ async def chat_unified(
                         pipeline_mgr = get_pipeline_manager()
                         pipeline_record = pipeline_mgr.get_pipeline(pipeline_id)
                         if pipeline_record and pipeline_record.get("components"):
-                            # Find first failed component
                             for comp in pipeline_record.get("components", []):
                                 if comp.get("hasError") or comp.get("status") == "failed":
                                     failed_component = {
@@ -1601,13 +1652,7 @@ async def chat_unified(
                 if failed_component:
                     api_data["failed_component"] = failed_component
                 
-                # Create error result for the response
-                error_result = {
-                    "error": str(e),
-                    "pipeline_id": pipeline_id,
-                    "status": "failed"
-                }
-                
+                # ✅ Return with pipeline_id and pipeline_action="failed"
                 return _assistant_response_payload(
                     chat_id=chat_id,
                     friendly_response=friendly,
@@ -1615,8 +1660,9 @@ async def chat_unified(
                     api_data=api_data,
                     state="initial",
                     exception=str(e),
-                    pipeline_result=error_result,  # NEW: Include error in pipeline_result
-                    pipeline_id=pipeline_id  # ✅ PASS pipeline_id HERE
+                    pipeline_result=error_result,
+                    pipeline_id=pipeline_id,
+                    pipeline_action="failed"  # ✅ ADD THIS
                 )
             finally:
                 try:
@@ -1726,13 +1772,16 @@ async def chat_unified(
                     ]
                 }
                 
+                # ✅ Return with pipeline_id and pipeline_action="created"
                 return _assistant_response_payload(
                     chat_id=chat_id,
                     friendly_response=friendly,
                     intent=intent_data,
                     api_data=api_data,
                     state="pipeline_proposed",
-                    output=output
+                    output=output,
+                    pipeline_id=pipeline_id,
+                    pipeline_action="created"  # ✅ ADD THIS
                 )
 
             except Exception as e:
@@ -1801,13 +1850,16 @@ async def chat_unified(
                         "modification": "edited"
                     }
                     
+                    # ✅ Return with pipeline_id and pipeline_action="created"
                     return _assistant_response_payload(
                         chat_id=chat_id,
                         friendly_response=friendly,
                         intent=intent_data,
                         api_data=api_data,
                         state="pipeline_proposed",
-                        output=output
+                        output=output,
+                        pipeline_id=pipeline_id,
+                        pipeline_action="created"  # ✅ ADD THIS
                     )
                 except Exception as e:
                     api_data = {
@@ -1930,8 +1982,6 @@ async def chat_unified_stream(
 
     def emit(obj: Dict[str, Any]) -> bytes:
         obj.setdefault("chat_id", chat_id)
-        # CHANGE: Fetch latest chat_name from session_manager to ensure it's up-to-date
-        # (e.g. if generated during the stream)
         current_session = session_manager.get_session(chat_id) or {}
         obj.setdefault("chat_name", current_session.get("chat_name"))
         obj.setdefault("state", current_session.get("state", "initial"))
@@ -2025,11 +2075,22 @@ async def chat_unified_stream(
                     f"- Type 'reject' or 'no' to cancel\n"
                     f"- Describe changes to modify the plan"
                 )
-                _add_and_mirror_message(chat_id, "assistant", friendly)
+                
+                # ✅ FIX: Add pipeline_id and pipeline_action to message
+                _add_and_mirror_message(
+                    chat_id, 
+                    "assistant", 
+                    friendly,
+                    pipeline_id=pipeline_id,
+                    pipeline_action="created"
+                )
+                
                 yield emit({
                     "type": "assistant_final", 
                     "content": friendly, 
                     "pipeline": pipeline,
+                    "pipeline_id": pipeline_id,  # ✅ Include in emit
+                    "pipeline_action": "created",
                     "output": {
                         "pipeline_id": pipeline_id,
                         "pipeline_name": pipeline_name,
@@ -2057,6 +2118,8 @@ async def chat_unified_stream(
             if intent_data["intent"] == "approval":
                 session_manager.update_session(chat_id, {"state": "executing"})
                 plan = session_local.get("proposed_pipeline", {})
+                pipeline_id = plan.get("pipeline_id")  # ✅ Get pipeline_id
+                
                 initial = (
                     f"✅ Approved! Starting execution of: **{plan.get('pipeline_name', 'pipeline')}**\n\n"
                     f"🚀 Processing, please wait...\n_(Using {plan.get('_generator', 'AI')} - {plan.get('_model', 'model')})_"
@@ -2069,8 +2132,6 @@ async def chat_unified_stream(
                 local_path, cleanup = download_to_temp_file(file_ref)
 
                 try:
-                    pipeline_id = plan.get("pipeline_id")
-                    
                     for event in execute_pipeline_streaming(
                         pipeline=plan,
                         file_path=local_path,
@@ -2117,7 +2178,16 @@ async def chat_unified_stream(
                             err = event.get("error", "Unknown error")
                             friendly_err = f"❌ Pipeline Failed\n\nError: {err}\n\nCompleted {len(steps_completed)} step(s) before failure."
                             session_manager.update_session(chat_id, {"state": "initial"})
-                            _add_and_mirror_message(chat_id, "assistant", friendly_err)
+                            
+                            # ✅ FIX: Add pipeline_id and pipeline_action to message
+                            _add_and_mirror_message(
+                                chat_id, 
+                                "assistant", 
+                                friendly_err,
+                                pipeline_id=pipeline_id,
+                                pipeline_action="failed",
+                                pipeline_result={"error": str(err), "status": "failed"}
+                            )
                             
                             # V3: Update status + result in S3
                             if pipeline_id:
@@ -2127,7 +2197,9 @@ async def chat_unified_stream(
                                 "type": "assistant_final", 
                                 "content": friendly_err, 
                                 "error": err,
-                                "exception": str(err)
+                                "exception": str(err),
+                                "pipeline_id": pipeline_id,
+                                "pipeline_action": "failed"
                             })
                             return
 
@@ -2136,14 +2208,14 @@ async def chat_unified_stream(
                         
                         # V3: Update pipeline status
                         if pipeline_id:
-                             _update_pipeline_status(pipeline_id, chat_id, "completed", result=final_payload)
-                             _record_model_attribution(
+                            _update_pipeline_status(pipeline_id, chat_id, "completed", result=final_payload)
+                            _record_model_attribution(
                                 pipeline_id=pipeline_id,
                                 session_id=chat_id,
                                 model_provider=plan.get("_model_provider", "unknown"),
                                 model_name=plan.get("_model", "unknown"),
                                 is_fallback=False
-                             )
+                            )
 
                         success_count = len([s for s in steps_completed if s.get("status") == "completed"])
                         result_text = _extract_user_facing_text(final_payload)
@@ -2155,15 +2227,28 @@ async def chat_unified_stream(
                             f"- Executor: {executor_used}\n\n"
                             f"{result_text or ''}"
                         )
-                        _add_and_mirror_message(chat_id, "assistant", friendly_final)
+                        
+                        # ✅ FIX: Add pipeline_id and pipeline_action to message
+                        _add_and_mirror_message(
+                            chat_id, 
+                            "assistant", 
+                            friendly_final,
+                            pipeline_id=pipeline_id,
+                            pipeline_action="executed",
+                            pipeline_result=final_payload
+                        )
+                        
                         yield emit({
                             "type": "assistant_final", 
                             "content": friendly_final, 
                             "result": final_payload,
+                            "pipeline_id": pipeline_id,
+                            "pipeline_action": "executed",
                             "output": {
                                 "component_summary": f"Executed {success_count} steps successfully",
                                 "steps_completed": success_count,
-                                "total_steps": len(steps_completed)
+                                "total_steps": len(steps_completed),
+                                "pipeline_id": pipeline_id
                             },
                             "final_output": {
                                 "text": result_text,
@@ -2175,18 +2260,42 @@ async def chat_unified_stream(
                         # Success but no payload?
                         done = f"✅ Pipeline Completed! Executed {len(steps_completed)} steps."
                         session_manager.update_session(chat_id, {"state": "initial"})
-                        _add_and_mirror_message(chat_id, "assistant", done)
+                        
+                        # ✅ FIX: Add pipeline_id and pipeline_action to message
+                        _add_and_mirror_message(
+                            chat_id, 
+                            "assistant", 
+                            done,
+                            pipeline_id=pipeline_id,
+                            pipeline_action="executed",
+                            pipeline_result={"message": "Completed without output", "status": "completed"}
+                        )
+                        
                         # V3 check
                         if pipeline_id:
                             _update_pipeline_status(pipeline_id, chat_id, "completed", result={"message": "Completed without output"})
                         
-                        yield emit({"type": "assistant_final", "content": done})
+                        yield emit({
+                            "type": "assistant_final", 
+                            "content": done,
+                            "pipeline_id": pipeline_id,
+                            "pipeline_action": "executed"
+                        })
                         return
 
                 except Exception as e:
                     friendly_err = f"❌ Pipeline Execution Failed\n\nError: {str(e)}"
                     session_manager.update_session(chat_id, {"state": "initial"})
-                    _add_and_mirror_message(chat_id, "assistant", friendly_err)
+                    
+                    # ✅ FIX: Add pipeline_id and pipeline_action to message
+                    _add_and_mirror_message(
+                        chat_id, 
+                        "assistant", 
+                        friendly_err,
+                        pipeline_id=pipeline_id,
+                        pipeline_action="failed",
+                        pipeline_result={"error": str(e), "status": "failed"}
+                    )
                     
                     # V3 Update
                     if pipeline_id:
@@ -2196,7 +2305,9 @@ async def chat_unified_stream(
                         "type": "assistant_final", 
                         "content": friendly_err, 
                         "error": str(e),
-                        "exception": str(e)
+                        "exception": str(e),
+                        "pipeline_id": pipeline_id,
+                        "pipeline_action": "failed"
                     })
                     return
                 finally:
@@ -2213,6 +2324,7 @@ async def chat_unified_stream(
                 return
 
             else:
+                # Edit request
                 try:
                     original_plan = session_local.get("proposed_pipeline", {})
                     edit_context = f"Original: {original_plan.get('pipeline_name')}. User wants: {message}"
@@ -2233,7 +2345,15 @@ async def chat_unified_stream(
                     session_manager.update_session(chat_id, {"proposed_pipeline": new_pipeline, "state": "pipeline_proposed"})
                     formatted = format_pipeline_for_display(new_pipeline)
                     friendly = formatted + f"\n\n```json\n{json.dumps(new_pipeline, indent=2)}\n```"
-                    _add_and_mirror_message(chat_id, "assistant", friendly)
+                    
+                    # ✅ FIX: Add pipeline_id and pipeline_action to message
+                    _add_and_mirror_message(
+                        chat_id, 
+                        "assistant", 
+                        friendly,
+                        pipeline_id=pipeline_id,
+                        pipeline_action="created"
+                    )
                     
                     # Get steps list for output
                     steps_list = new_pipeline.get("pipeline_steps", new_pipeline.get("components", []))
@@ -2242,6 +2362,8 @@ async def chat_unified_stream(
                         "type": "assistant_final", 
                         "content": friendly, 
                         "pipeline": new_pipeline,
+                        "pipeline_id": pipeline_id,
+                        "pipeline_action": "created",
                         "output": {
                             "pipeline_id": pipeline_id,
                             "pipeline_name": new_pipeline.get("pipeline_name", "Document Processing"),