Update api_routes_v2.py

api_routes_v2.py

@@ -421,6 +421,9 @@ from services.pipeline_executor import execute_pipeline_streaming, execute_pipel
 from services.session_manager import session_manager
 from services.intent_classifier import intent_classifier
 
+# Add or update this import near the top (with other imports)
+from datetime import datetime, timedelta
+
 router = APIRouter(prefix="/api/v2", tags=["MasterLLM API V2 - Enhanced"])
 
 # ========================
@@ -468,6 +471,199 @@ class ChatResponse(BaseModel):
     fileUrl: Optional[str] = None
 
 
+
+## helpers for presigned url chat name and some more updates
+
+def _resolve_bedrock_model_for_titles(session: Dict[str, Any]) -> str:
+    """
+    CHANGE: NEW helper.
+    Use the same model as other tasks if available (session.proposed_pipeline._model).
+    Fallback to env, then a sane default.
+    """
+    try:
+        model = (session.get("proposed_pipeline") or {}).get("_model")
+    except Exception:
+        model = None
+    if not model:
+        model = os.getenv("BEDROCK_MODEL_ID") or os.getenv("BEDROCK_DEFAULT_MODEL") or "anthropic.claude-3-haiku-20240307"
+    return model
+
+
+def _bedrock_invoke_title(model_id: str, prompt_text: str) -> str:
+    """
+    CHANGE: NEW helper.
+    Minimal Bedrock invocation for Anthropic/Titan models to produce a short title.
+    If invocation fails, returns 'New Chat'.
+    """
+    try:
+        bedrock_runtime = boto3.client("bedrock-runtime", region_name=AWS_REGION)
+        if model_id.startswith("anthropic."):
+            # Anthropic Messages on Bedrock
+            body = {
+                "anthropic_version": "bedrock-2023-05-31",
+                "max_tokens": 48,
+                "temperature": 0.2,
+                "messages": [{"role": "user", "content": [{"type": "text", "text": prompt_text}]}],
+            }
+            resp = bedrock_runtime.invoke_model(
+                modelId=model_id,
+                accept="application/json",
+                contentType="application/json",
+                body=json.dumps(body),
+            )
+            payload = json.loads(resp["body"].read())
+            # Extract first text piece
+            text = ""
+            if isinstance(payload.get("content"), list) and payload["content"]:
+                part = payload["content"][0]
+                if isinstance(part, dict):
+                    text = part.get("text") or ""
+            return (text or "").strip().strip('"').strip() or "New Chat"
+        else:
+            # Titan text models (or similar)
+            body = {
+                "inputText": prompt_text,
+                "textGenerationConfig": {"maxTokenCount": 64, "temperature": 0.2, "stopSequences": []},
+            }
+            resp = bedrock_runtime.invoke_model(
+                modelId=model_id,
+                accept="application/json",
+                contentType="application/json",
+                body=json.dumps(body),
+            )
+            payload = json.loads(resp["body"].read())
+            results = payload.get("results") or []
+            if results:
+                return (results[0].get("outputText") or "").strip().strip('"').strip() or "New Chat"
+            return "New Chat"
+    except Exception:
+        return "New Chat"
+
+
+def _maybe_generate_chat_name(chat_id: str):
+    """
+    CHANGE: NEW helper.
+    Auto-generate a succinct chat title after the first real user message (not the 'Uploaded file:' stub).
+    Uses the same Bedrock model as used elsewhere where possible.
+    """
+    try:
+        s = session_manager.get_session(chat_id) or {}
+        if s.get("chat_name"):
+            return
+        msgs = list(s.get("messages", []))
+        first_user = None
+        for m in msgs:
+            if (m.get("role") or "") == "user":
+                content = (m.get("content") or "").strip()
+                if not content.lower().startswith("uploaded file:"):
+                    first_user = content
+                    break
+        if not first_user:
+            return
+        file_name = (s.get("file_metadata") or {}).get("file_name")
+        prompt = (
+            "Create a succinct, descriptive 3–6 word title for this chat session based on the first user message.\n"
+            "Return only the title, without quotes.\n\n"
+            f"First message: {first_user}\n"
+            f"File name (optional): {file_name or 'N/A'}"
+        )
+        model_id = _resolve_bedrock_model_for_titles(s)
+        title = _bedrock_invoke_title(model_id, prompt) or "New Chat"
+        session_manager.update_session(
+            chat_id,
+            {
+                "chat_name": title[:100],
+                "chat_name_generated_at": datetime.utcnow().isoformat() + "Z",
+                "chat_name_model": model_id,
+            },
+        )
+    except Exception:
+        # Do not block flow on errors
+        pass
+
+
+def _generate_presigned_get_url(bucket: str, key: str, expires_in: int = 604800) -> Dict[str, str]:
+    """
+    CHANGE: NEW helper.
+    Generate a presigned S3 GET URL with max expiry (7 days). We only generate once on upload.
+    """
+    try:
+        url = s3.generate_presigned_url(
+            "get_object", Params={"Bucket": bucket, "Key": key}, ExpiresIn=expires_in
+        )
+        expires_at = (datetime.utcnow() + timedelta(seconds=expires_in)).isoformat() + "Z"
+        return {"presigned_url": url, "presigned_expires_at": expires_at}
+    except Exception:
+        return {}
+
+
+def _push_pipeline_history(chat_id: str, pipeline: Dict[str, Any], created_from: str = "request", message: Optional[str] = None):
+    """
+    CHANGE: NEW helper.
+    Prepend a pipeline record to session.pipelines_history (unbounded, latest first).
+    """
+    try:
+        s = session_manager.get_session(chat_id) or {}
+        hist = list(s.get("pipelines_history", []))
+        entry = {
+            "pipeline": pipeline,
+            "created_at": datetime.utcnow().isoformat() + "Z",
+            "created_from": created_from,
+            "message": message,
+            "status": "proposed",
+        }
+        hist.insert(0, entry)
+        session_manager.update_session(chat_id, {"pipelines_history": hist})
+    except Exception:
+        pass
+
+
+def _mark_latest_pipeline_history_status(chat_id: str, status: str, result_preview: Optional[str] = None):
+    """
+    CHANGE: NEW helper.
+    Update the most recent pipelines_history entry with status and optional result preview.
+    """
+    try:
+        s = session_manager.get_session(chat_id) or {}
+        hist = list(s.get("pipelines_history", []))
+        if not hist:
+            return
+        hist[0]["status"] = status
+        if result_preview is not None:
+            hist[0]["result_preview"] = result_preview[:500]
+        session_manager.update_session(chat_id, {"pipelines_history": hist})
+    except Exception:
+        pass
+
+
+def _extract_user_facing_text(obj: Any) -> str:
+    """
+    CHANGE: NEW helper.
+    Heuristically extract user-facing text from pipeline results.
+    """
+    try:
+        if isinstance(obj, str):
+            return obj
+        if isinstance(obj, dict):
+            for k in ["summary", "final_text", "content", "text", "output"]:
+                v = obj.get(k)
+                if isinstance(v, str) and v.strip():
+                    return v.strip()
+                if isinstance(v, list):
+                    texts = [x for x in v if isinstance(x, str)]
+                    if texts:
+                        return "\n".join(texts[:3]).strip()
+            return json.dumps(obj, ensure_ascii=False)[:2000]
+        if isinstance(obj, list):
+            texts = [x for x in obj if isinstance(x, str)]
+            if texts:
+                return "\n".join(texts[:5])
+            return json.dumps(obj, ensure_ascii=False)[:2000]
+        return ""
+    except Exception:
+        return ""
+
+
 # ========================
 # HELPERS
 # ========================
@@ -532,10 +728,31 @@ def _normalize_history_for_api(chat_id: str) -> List[Message]:
     return history
 
 
+# def _add_and_mirror_message(chat_id: str, role: str, content: str):
+#     """
+#     Add a message via session_manager AND mirror it into the session doc
+#     so history is always available in responses.
+#     """
+#     session_manager.add_message(chat_id, role, content)
+#     try:
+#         s = session_manager.get_session(chat_id) or {}
+#         msgs = list(s.get("messages", []))
+#         msgs.append({
+#             "role": role,
+#             "content": content if isinstance(content, str) else json.dumps(content, ensure_ascii=False),
+#             "timestamp": datetime.utcnow().isoformat() + "Z",
+#         })
+#         session_manager.update_session(chat_id, {"messages": msgs})
+#     except Exception:
+#         # Don't block the flow on mirror issues
+#         pass
+
 def _add_and_mirror_message(chat_id: str, role: str, content: str):
     """
     Add a message via session_manager AND mirror it into the session doc
     so history is always available in responses.
+    CHANGE: Now we always include file_data: {"has_file": False} for every message by default,
+    so Message.file is present per message in history.
     """
     session_manager.add_message(chat_id, role, content)
     try:
@@ -545,6 +762,7 @@ def _add_and_mirror_message(chat_id: str, role: str, content: str):
             "role": role,
             "content": content if isinstance(content, str) else json.dumps(content, ensure_ascii=False),
             "timestamp": datetime.utcnow().isoformat() + "Z",
+            "file_data": {"has_file": False}  # CHANGE: ensure explicit boolean on every message
         })
         session_manager.update_session(chat_id, {"messages": msgs})
     except Exception:
@@ -634,6 +852,9 @@ def upload_stream_to_s3(chat_id: str, file: UploadFile) -> str:
     """
     Stream an UploadFile directly to S3, return s3:// URI.
     Supports optional SSE via env S3_SSE and S3_KMS_KEY_ID.
+    CHANGE:
+    - Generate a presigned GET URL with max expiry (7 days) once at upload time.
+    - Store presigned_url and presigned_expires_at in session.file_metadata (do not regenerate later).
     """
     key = f"{S3_PREFIX}/{chat_id}/{file.filename}"
     config = TransferConfig(multipart_threshold=8 * 1024 * 1024, max_concurrency=4)
@@ -665,8 +886,10 @@ def upload_stream_to_s3(chat_id: str, file: UploadFile) -> str:
         )
 
     s3_uri = f"s3://{S3_BUCKET}/{key}"
-    
-    # Store file metadata in session
+    # CHANGE: generate and store a single presigned URL (max 7 days) for later reuse
+    presigned = _generate_presigned_get_url(S3_BUCKET, key, expires_in=604800)
+
+    # Store file metadata in session (adds presigned fields)
     session_manager.update_session(chat_id, {
         "current_file": s3_uri,
         "state": "initial",
@@ -674,10 +897,11 @@ def upload_stream_to_s3(chat_id: str, file: UploadFile) -> str:
             "has_file": True,
             "file_name": file.filename,
             "file_url": s3_uri,
-            "uploaded_at": datetime.utcnow().isoformat() + "Z"
+            "uploaded_at": datetime.utcnow().isoformat() + "Z",
+            **presigned  # CHANGE
         }
     })
-    
+
     # Create a user message with file metadata (instead of system message)
     file_message = {
         "role": "user",
@@ -689,13 +913,13 @@ def upload_stream_to_s3(chat_id: str, file: UploadFile) -> str:
             "file_url": s3_uri
         }
     }
-    
+
     # Get existing messages and add the file message
     s = session_manager.get_session(chat_id) or {}
     msgs = list(s.get("messages", []))
     msgs.append(file_message)
     session_manager.update_session(chat_id, {"messages": msgs})
-    
+
     return s3_uri
 
 
@@ -711,25 +935,19 @@ async def get_all_sessions(
 ):
     """
     Get all session IDs from the database with optional pagination
-    
-    Args:
-        limit: Maximum number of sessions to return (default: 100)
-        skip: Number of sessions to skip (for pagination)
-        include_stats: If True, include session statistics
-    
-    Returns:
-        List of sessions with basic info or full stats
+
+    CHANGE: When include_stats=True, include 'chat_name' (added field only).
     """
     try:
         all_session_ids = session_manager.get_all_session_ids()
-        
+
         if not all_session_ids:
             return {"sessions": [], "pagination": {"total": 0, "returned": 0}}
-        
+
         # Apply pagination
         total_sessions = len(all_session_ids)
         paginated_ids = all_session_ids[skip:skip + limit]
-        
+
         if not include_stats:
             # Return just session IDs with pagination info
             sessions_basic = [
@@ -740,7 +958,7 @@ async def get_all_sessions(
                 }
                 for sid in paginated_ids
             ]
-            
+
             return {
                 "sessions": sessions_basic,
                 "pagination": {
@@ -751,7 +969,7 @@ async def get_all_sessions(
                     "has_more": total_sessions > (skip + limit)
                 }
             }
-        
+
         # Include detailed statistics for each session
         sessions_with_stats = []
         for session_id in paginated_ids:
@@ -760,12 +978,12 @@ async def get_all_sessions(
                 # Format datetime objects for JSON serialization
                 created_at = session.get("created_at")
                 last_activity = session.get("last_activity")
-                
+
                 if isinstance(created_at, datetime):
                     created_at = created_at.isoformat()
                 if isinstance(last_activity, datetime):
                     last_activity = last_activity.isoformat()
-                
+
                 sessions_with_stats.append({
                     "session_id": session_id,
                     "user_id": session.get("user_id"),
@@ -773,11 +991,12 @@ async def get_all_sessions(
                     "last_activity": last_activity,
                     "state": session.get("state", "unknown"),
                     "current_file": session.get("current_file"),
+                    "chat_name": session.get("chat_name"),  # CHANGE: added field
                     "stats": session.get("stats", {}),
                     "total_messages": len(session.get("conversation_history", [])),
                     "pipeline_executions_count": len(session.get("pipeline_executions", []))
                 })
-        
+
         return {
             "sessions": sessions_with_stats,
             "pagination": {
@@ -788,7 +1007,7 @@ async def get_all_sessions(
                 "has_more": total_sessions > (skip + limit)
             }
         }
-    
+
     except Exception as e:
         raise HTTPException(
             status_code=500,
@@ -807,18 +1026,13 @@ async def get_session_history(
 ):
     """
     Get conversation history for a session
-    
-    Args:
-        session_id: Session identifier
-        limit: Maximum number of messages to return (default: 50)
-    
-    Returns:
-        Session history including session ID
+
+    CHANGE: Adds 'chat_name' and 'pipelines_history' (added fields only).
     """
     try:
         # Get the history using session manager
         history = session_manager.get_session_history(session_id, limit)
-        
+
         # Format datetime objects to ISO strings for JSON serialization
         formatted_history = []
         for msg in history:
@@ -826,15 +1040,19 @@ async def get_session_history(
             if "timestamp" in msg_copy and isinstance(msg_copy["timestamp"], datetime):
                 msg_copy["timestamp"] = msg_copy["timestamp"].isoformat()
             formatted_history.append(msg_copy)
-        
+
+        s = session_manager.get_session(session_id) or {}
+
         # Return with session_id included
         return {
             "session_id": session_id,
             "history": formatted_history,
             "count": len(formatted_history),
-            "limit": limit
+            "limit": limit,
+            "chat_name": s.get("chat_name"),                    # CHANGE
+            "pipelines_history": s.get("pipelines_history", []) # CHANGE
         }
-    
+
     except Exception as e:
         raise HTTPException(
             status_code=500,
@@ -861,6 +1079,13 @@ async def chat_unified(
     - Handles casual chat, pipeline request, approve/reject, and edits.
     - On approval, executes the pipeline (non-stream) and returns the final result.
     Returns assistant_response + full history (role/content).
+
+    CHANGE SUMMARY:
+    - Generate session name after first real user message via Bedrock LLM.
+    - Include presigned_url in file ack when a file is uploaded here (from stored metadata; do not regenerate).
+    - Maintain unbounded pipelines_history (latest first) on pipeline create/edit.
+    - Final output after pipeline execution includes actual user-facing result (summary) instead of generic text.
+    - No response field names changed; only fields were added where applicable.
     """
 
     # Support JSON payloads too
@@ -886,13 +1111,21 @@ async def chat_unified(
     file_info = None
     if file is not None:
         s3_uri = upload_stream_to_s3(chat_id, file)
-        file_info = {"bucket": S3_BUCKET, "key": s3_uri.split(f"s3://{S3_BUCKET}/", 1)[1], "s3_uri": s3_uri}
+        # CHANGE: Include presigned fields from saved metadata (no regeneration)
+        meta = (session_manager.get_session(chat_id) or {}).get("file_metadata", {}) or {}
+        file_info = {
+            "bucket": S3_BUCKET,
+            "key": s3_uri.split(f"s3://{S3_BUCKET}/", 1)[1],
+            "s3_uri": s3_uri,
+            "presigned_url": meta.get("presigned_url"),               # CHANGE
+            "presigned_expires_at": meta.get("presigned_expires_at")  # CHANGE
+        }
         session = _get_session_or_init(chat_id)
 
     # If no message and only a file was sent, respond with an acknowledgement
     if (message is None or str(message).strip() == "") and file_info:
         friendly = "📁 File uploaded successfully. Tell me what you'd like to do with it (e.g., extract text, get tables, summarize)."
-        api_data = {"type": "file_uploaded", "file": file_info, "next_action": "send_instruction"}
+        api_data = {"type": "file_uploaded", "file": file_info, "next_action": "send_instruction"}  # CHANGE: file contains presigned
         return _assistant_response_payload(chat_id, friendly, {"intent": "file_uploaded"}, api_data, session.get("state", "initial"))
 
     # If still no message, nudge the user
@@ -903,6 +1136,8 @@ async def chat_unified(
 
     # Add user message
     _add_and_mirror_message(chat_id, "user", message)
+    # CHANGE: maybe generate a session title after first real user message
+    _maybe_generate_chat_name(chat_id)
 
     # Classify intent
     intent_data = intent_classifier.classify_intent(message)
@@ -964,10 +1199,11 @@ async def chat_unified(
                     prefer_bedrock=bool(prefer_bedrock),
                 )
                 session_manager.update_session(chat_id, {"pipeline_result": result, "state": "initial"})
-                friendly = (
-                    f"🎉 Pipeline completed successfully!\n"
-                    f"✅ All done! What else would you like me to help you with?"
-                )
+                # CHANGE: include actual user-facing result
+                result_text = _extract_user_facing_text(result)
+                preview = (result_text or json.dumps(result, ensure_ascii=False))[:500]
+                _mark_latest_pipeline_history_status(chat_id, "executed", result_preview=preview)  # CHANGE
+                friendly = f"🎉 Pipeline completed successfully!\n\n{result_text or '✅ All done!'}"  # CHANGE
                 api_data = {
                     "type": "pipeline_completed",
                     "result": result,
@@ -976,6 +1212,7 @@ async def chat_unified(
                 return _assistant_response_payload(chat_id, friendly, {"intent": "pipeline_execute"}, api_data, "initial")
             except Exception as e:
                 session_manager.update_session(chat_id, {"state": "initial"})
+                _mark_latest_pipeline_history_status(chat_id, "failed", result_preview=str(e))  # CHANGE
                 friendly = f"❌ Pipeline execution failed: {str(e)}"
                 api_data = {"type": "error", "error_code": "PIPELINE_EXECUTION_FAILED", "message": str(e)}
                 return _assistant_response_payload(chat_id, friendly, {"intent": "pipeline_execute"}, api_data, "initial")
@@ -1015,6 +1252,7 @@ async def chat_unified(
                 )
 
                 session_manager.update_session(chat_id, {"proposed_pipeline": pipeline, "state": "pipeline_proposed"})
+                _push_pipeline_history(chat_id, pipeline, created_from="request", message=message)  # CHANGE
 
                 pipeline_name = pipeline.get("pipeline_name", "Document Processing")
                 steps_list = pipeline.get("pipeline_steps", [])
@@ -1074,6 +1312,7 @@ async def chat_unified(
                         prefer_bedrock=bool(prefer_bedrock)
                     )
                     session_manager.update_session(chat_id, {"proposed_pipeline": new_pipeline, "state": "pipeline_proposed"})
+                    _push_pipeline_history(chat_id, new_pipeline, created_from="edit", message=message)  # CHANGE
 
                     formatted = format_pipeline_for_display(new_pipeline)
                     friendly = formatted + f"\n\n```json\n{json.dumps(new_pipeline, indent=2)}\n```"
@@ -1147,6 +1386,12 @@ async def chat_unified_stream(
     - Accepts multipart/form-data (file + message) OR JSON.
     - Uploads file if included.
     - On approval, streams execution progress and final result.
+
+    CHANGE SUMMARY:
+    - Generate session name after first real user message via Bedrock LLM.
+    - Include presigned_url in file ack when a file is uploaded here (from stored metadata; do not regenerate).
+    - Maintain pipelines_history on pipeline create/edit, and mark executed/failed after run.
+    - Stream final includes actual user-facing result text.
     """
 
     # Parse JSON if needed
@@ -1172,7 +1417,14 @@ async def chat_unified_stream(
     uploaded_file_info = None
     if file is not None:
         s3_uri = upload_stream_to_s3(chat_id, file)
-        uploaded_file_info = {"bucket": S3_BUCKET, "key": s3_uri.split(f"s3://{S3_BUCKET}/", 1)[1], "s3_uri": s3_uri}
+        meta = (session_manager.get_session(chat_id) or {}).get("file_metadata", {}) or {}
+        uploaded_file_info = {
+            "bucket": S3_BUCKET,
+            "key": s3_uri.split(f"s3://{S3_BUCKET}/", 1)[1],
+            "s3_uri": s3_uri,
+            "presigned_url": meta.get("presigned_url"),               # CHANGE
+            "presigned_expires_at": meta.get("presigned_expires_at")  # CHANGE
+        }
         session = _get_session_or_init(chat_id)
 
     def emit(obj: Dict[str, Any]) -> bytes:
@@ -1199,6 +1451,8 @@ async def chat_unified_stream(
 
         # Add user message
         _add_and_mirror_message(chat_id, "user", message)
+        # CHANGE: Maybe generate chat title
+        _maybe_generate_chat_name(chat_id)
 
         # Classify
         intent_data = intent_classifier.classify_intent(message)
@@ -1237,6 +1491,7 @@ async def chat_unified_stream(
                     prefer_bedrock=bool(prefer_bedrock),
                 )
                 session_manager.update_session(chat_id, {"proposed_pipeline": pipeline, "state": "pipeline_proposed"})
+                _push_pipeline_history(chat_id, pipeline, created_from="request", message=message)  # CHANGE
 
                 pipeline_name = pipeline.get("pipeline_name", "Document Processing")
                 steps_list = pipeline.get("pipeline_steps", [])
@@ -1323,6 +1578,7 @@ async def chat_unified_stream(
                             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)
+                            _mark_latest_pipeline_history_status(chat_id, "failed", result_preview=str(err))  # CHANGE
                             yield emit({"type": "assistant_final", "content": friendly_err, "error": err, "history": [m.dict() for m in _normalize_history_for_api(chat_id)]})
                             return
 
@@ -1336,13 +1592,17 @@ async def chat_unified_stream(
                             executor=executor_used
                         )
                         success_count = len([s for s in steps_completed if s.get("status") == "completed"])
+                        # CHANGE: include actual user-facing result
+                        result_text = _extract_user_facing_text(final_payload)
+                        preview = (result_text or json.dumps(final_payload, ensure_ascii=False))[:500]
+                        _mark_latest_pipeline_history_status(chat_id, "executed", result_preview=preview)  # CHANGE
                         friendly_final = (
                             f"🎉 Pipeline Completed Successfully!\n"
                             f"- Pipeline: {plan.get('pipeline_name', 'Document Processing')}\n"
                             f"- Total Steps: {len(steps_completed)}\n"
                             f"- Successful: {success_count}\n"
-                            f"- Executor: {executor_used}\n"
-                            f"✅ All done! What else would you like me to help you with?"
+                            f"- Executor: {executor_used}\n\n"
+                            f"{result_text or ''}"
                         )
                         _add_and_mirror_message(chat_id, "assistant", friendly_final)
                         yield emit({"type": "assistant_final", "content": friendly_final, "result": final_payload, "history": [m.dict() for m in _normalize_history_for_api(chat_id)]})
@@ -1358,6 +1618,7 @@ async def chat_unified_stream(
                     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)
+                    _mark_latest_pipeline_history_status(chat_id, "failed", result_preview=str(e))  # CHANGE
                     yield emit({"type": "assistant_final", "content": friendly_err, "error": str(e), "history": [m.dict() for m in _normalize_history_for_api(chat_id)]})
                     return
                 finally:
@@ -1383,6 +1644,7 @@ async def chat_unified_stream(
                         prefer_bedrock=bool(prefer_bedrock)
                     )
                     session_manager.update_session(chat_id, {"proposed_pipeline": new_pipeline, "state": "pipeline_proposed"})
+                    _push_pipeline_history(chat_id, new_pipeline, created_from="edit", message=message)  # CHANGE
                     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)
@@ -1412,6 +1674,11 @@ async def smart_chat(request: ChatRequest):
     """
     Kept for compatibility with existing clients (non-stream).
     For a single all-in-one endpoint, use /api/v2/chat/unified.
+
+    CHANGE SUMMARY:
+    - Generate session name after first real user message.
+    - Maintain pipelines_history on pipeline create/edit.
+    - (This legacy endpoint does not execute on approval; unchanged to avoid breaking behavior.)
     """
     chat_id = _ensure_chat(request.chat_id)
     session = _get_session_or_init(chat_id)
@@ -1421,6 +1688,8 @@ async def smart_chat(request: ChatRequest):
         session = _get_session_or_init(chat_id)
 
     _add_and_mirror_message(chat_id, "user", request.message)
+    # CHANGE: maybe name the chat
+    _maybe_generate_chat_name(chat_id)
 
     intent_data = intent_classifier.classify_intent(request.message)
     current_state = session.get("state", "initial")
@@ -1504,6 +1773,7 @@ async def smart_chat(request: ChatRequest):
                 )
 
                 session_manager.update_session(chat_id, {"proposed_pipeline": pipeline, "state": "pipeline_proposed"})
+                _push_pipeline_history(chat_id, pipeline, created_from="request", message=request.message)  # CHANGE
 
                 pipeline_name = pipeline.get("pipeline_name", "Document Processing")
                 steps_list = pipeline.get("pipeline_steps", [])
@@ -1562,6 +1832,7 @@ async def smart_chat(request: ChatRequest):
                         prefer_bedrock=request.prefer_bedrock
                     )
                     session_manager.update_session(chat_id, {"proposed_pipeline": new_pipeline, "state": "pipeline_proposed"})
+                    _push_pipeline_history(chat_id, new_pipeline, created_from="edit", message=request.message)  # CHANGE
 
                     formatted = format_pipeline_for_display(new_pipeline)
                     friendly = formatted + f"\n\n```json\n{json.dumps(new_pipeline, indent=2)}\n```"
@@ -1621,6 +1892,12 @@ def smart_chat_stream(request: ChatRequest):
     """
     Kept for compatibility (streaming NDJSON).
     For the all-in-one streaming flow, use /api/v2/chat/unified/stream.
+
+    CHANGE SUMMARY:
+    - Generate session name after first real user message.
+    - Maintain pipelines_history on pipeline create/edit.
+    - Mark latest pipelines_history as executed/failed when streaming completes.
+    - Final streamed assistant message includes user-facing result text.
     """
 
     def gen() -> Generator[bytes, None, None]:
@@ -1632,6 +1909,8 @@ def smart_chat_stream(request: ChatRequest):
             session = _get_session_or_init(chat_id)
 
         _add_and_mirror_message(chat_id, "user", request.message)
+        # CHANGE: maybe generate chat name
+        _maybe_generate_chat_name(chat_id)
 
         intent_data = intent_classifier.classify_intent(request.message)
         current_state = session.get("state", "initial")
@@ -1673,6 +1952,7 @@ def smart_chat_stream(request: ChatRequest):
                     prefer_bedrock=request.prefer_bedrock
                 )
                 session_manager.update_session(chat_id, {"proposed_pipeline": pipeline, "state": "pipeline_proposed"})
+                _push_pipeline_history(chat_id, pipeline, created_from="request", message=request.message)  # CHANGE
 
                 pipeline_name = pipeline.get("pipeline_name", "Document Processing")
                 steps_list = pipeline.get("pipeline_steps", [])
@@ -1759,9 +2039,10 @@ def smart_chat_stream(request: ChatRequest):
 
                         elif etype == "error":
                             err = event.get("error", "Unknown error")
-                            friendly_err = f"❌ Pipeline Failed\n\nError: {err}\n\nCompleted {len(steps_completed)} step(s) before failure."
+                            friendly_err = f"❌ Pipeline Execution 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)
+                            _mark_latest_pipeline_history_status(chat_id, "failed", result_preview=str(err))  # CHANGE
                             yield emit({"type": "assistant_final", "content": friendly_err, "error": err, "history": [m.dict() for m in _normalize_history_for_api(chat_id)]})
                             return
 
@@ -1775,13 +2056,16 @@ def smart_chat_stream(request: ChatRequest):
                             executor=executor_used
                         )
                         success_count = len([s for s in steps_completed if s.get("status") == "completed"])
+                        result_text = _extract_user_facing_text(final_payload)  # CHANGE
+                        preview = (result_text or json.dumps(final_payload, ensure_ascii=False))[:500]
+                        _mark_latest_pipeline_history_status(chat_id, "executed", result_preview=preview)  # CHANGE
                         friendly_final = (
                             f"🎉 Pipeline Completed Successfully!\n"
                             f"- Pipeline: {plan.get('pipeline_name', 'Document Processing')}\n"
                             f"- Total Steps: {len(steps_completed)}\n"
                             f"- Successful: {success_count}\n"
-                            f"- Executor: {executor_used}\n"
-                            f"✅ All done! What else would you like me to help you with?"
+                            f"- Executor: {executor_used}\n\n"
+                            f"{result_text or ''}"
                         )
                         _add_and_mirror_message(chat_id, "assistant", friendly_final)
                         yield emit({"type": "assistant_final", "content": friendly_final, "result": final_payload, "history": [m.dict() for m in _normalize_history_for_api(chat_id)]})
@@ -1797,6 +2081,7 @@ def smart_chat_stream(request: ChatRequest):
                     friendly_err = f"❌ Pipeline Execution Failed\n\nError: {str(e)}\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)
+                    _mark_latest_pipeline_history_status(chat_id, "failed", result_preview=str(e))  # CHANGE
                     yield emit({"type": "assistant_final", "content": friendly_err, "error": str(e), "history": [m.dict() for m in _normalize_history_for_api(chat_id)]})
                     return
                 finally:
@@ -1822,6 +2107,7 @@ def smart_chat_stream(request: ChatRequest):
                         prefer_bedrock=request.prefer_bedrock
                     )
                     session_manager.update_session(chat_id, {"proposed_pipeline": new_pipeline, "state": "pipeline_proposed"})
+                    _push_pipeline_history(chat_id, new_pipeline, created_from="edit", message=request.message)  # CHANGE
                     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)
@@ -1846,6 +2132,10 @@ def smart_chat_stream(request: ChatRequest):
 
 @router.post("/chats/{chat_id}/pipeline/execute", response_model=ChatResponse)
 async def execute_pipeline_now(chat_id: str):
+    """
+    CHANGE: After execution, assistant_response includes user-facing result text.
+    Also mark latest pipelines_history entry as executed/failed.
+    """
     session = session_manager.get_session(chat_id)
     if not session:
         raise HTTPException(status_code=404, detail="Chat not found")
@@ -1865,11 +2155,16 @@ async def execute_pipeline_now(chat_id: str):
             prefer_bedrock=True
         )
         session_manager.update_session(chat_id, {"pipeline_result": result, "state": "initial"})
-        friendly = "🎉 Pipeline completed. Ready for your next task!"
+        # CHANGE: include actual result text
+        result_text = _extract_user_facing_text(result)
+        preview = (result_text or json.dumps(result, ensure_ascii=False))[:500]
+        _mark_latest_pipeline_history_status(chat_id, "executed", result_preview=preview)  # CHANGE
+        friendly = f"🎉 Pipeline completed.\n\n{result_text or ''}"  # CHANGE
         api_data = {"type": "pipeline_completed", "result": result, "pipeline": plan}
         return _assistant_response_payload(chat_id, friendly, {"intent": "pipeline_execute"}, api_data, "initial")
     except Exception as e:
         session_manager.update_session(chat_id, {"state": "initial"})
+        _mark_latest_pipeline_history_status(chat_id, "failed", result_preview=str(e))  # CHANGE
         friendly = f"❌ Pipeline execution failed: {str(e)}"
         api_data = {"type": "error", "error_code": "PIPELINE_EXECUTION_FAILED", "message": str(e)}
         return _assistant_response_payload(chat_id, friendly, {"intent": "pipeline_execute"}, api_data, "initial")
@@ -1932,6 +2227,9 @@ def create_chat():
 
 @router.get("/chats/{chat_id}")
 def get_chat(chat_id: str):
+    """
+    CHANGE: Add chat_name and pipelines_history to session retrieval (added fields only).
+    """
     s = session_manager.get_session(chat_id)
     if not s:
         raise HTTPException(status_code=404, detail="Chat not found")
@@ -1941,6 +2239,8 @@ def get_chat(chat_id: str):
         "current_file": s.get("current_file"),
         "created_at": s.get("created_at"),
         "updated_at": s.get("updated_at"),
+        "chat_name": s.get("chat_name"),                  # CHANGE
+        "pipelines_history": s.get("pipelines_history", [])  # CHANGE
     }
 
 
@@ -1955,32 +2255,28 @@ def get_chat(chat_id: str):
 def get_chat_history(chat_id: str):
     """
     Get conversation history for a specific chat
-    
-    Args:
-        chat_id: Chat identifier
-    
-    Returns:
-        Chat history with session ID and file metadata
+
+    CHANGE: Add chat_name and pipelines_history into the response (added fields only).
     """
     s = session_manager.get_session(chat_id)
     if not s:
         raise HTTPException(status_code=404, detail="Chat not found")
-    
+
     # Get the normalized history (system messages filtered out)
     history = [m.dict() for m in _normalize_history_for_api(chat_id)]
-    
+
     # Get file metadata
     file_metadata = s.get("file_metadata", {})
-    
+
     # Format datetime objects for JSON serialization
     created_at = s.get("created_at")
     updated_at = s.get("updated_at")
-    
+
     if isinstance(created_at, datetime):
         created_at = created_at.isoformat()
     if isinstance(updated_at, datetime):
         updated_at = updated_at.isoformat()
-    
+
     # Return with session_id and file metadata
     return {
         "session_id": chat_id,
@@ -1992,7 +2288,9 @@ def get_chat_history(chat_id: str):
         "updated_at": updated_at,
         "file": file_metadata.get("has_file", False),
         "fileName": file_metadata.get("file_name"),
-        "fileUrl": file_metadata.get("file_url")
+        "fileUrl": file_metadata.get("file_url"),
+        "chat_name": s.get("chat_name"),                       # CHANGE
+        "pipelines_history": s.get("pipelines_history", [])    # CHANGE
     }
 
 
@@ -2017,12 +2315,23 @@ async def send_message_to_chat(chat_id: str, payload: ChatRequest):
 
 @router.post("/chats/{chat_id}/upload")
 async def upload_file_to_chat(chat_id: str, file: UploadFile = File(...)):
+    """
+    CHANGE: Include presigned_url and presigned_expires_at in response file object.
+    These are generated once during upload_stream_to_s3 and stored in session; we reuse them here.
+    """
     chat_id = _ensure_chat(chat_id)
     s3_uri = upload_stream_to_s3(chat_id, file)
+    meta = (session_manager.get_session(chat_id) or {}).get("file_metadata", {}) or {}  # CHANGE
     return {
         "status": "success",
         "message": "File uploaded to S3",
-        "file": {"bucket": S3_BUCKET, "key": s3_uri.split(f"s3://{S3_BUCKET}/", 1)[1], "s3_uri": s3_uri},
+        "file": {
+            "bucket": S3_BUCKET,
+            "key": s3_uri.split(f"s3://{S3_BUCKET}/", 1)[1],
+            "s3_uri": s3_uri,
+            "presigned_url": meta.get("presigned_url"),                 # CHANGE
+            "presigned_expires_at": meta.get("presigned_expires_at")    # CHANGE
+        },
         "chat_id": chat_id,
         "next_action": "💬 Now tell me what you'd like to do with this document"
     }