Fix role propagation in ingestion pipeline and improve error handling

app.py

@@ -2,9 +2,12 @@ import gradio as gr
 import requests
 import json
 import os
+import sys
 from pathlib import Path
 from collections import Counter
 from datetime import datetime
+from dotenv import load_dotenv
+load_dotenv()
 
 try:
     import plotly.graph_objects as go
@@ -334,9 +337,21 @@ def ingest_document(
     doc_id: str,
     metadata_json: str
 ):
+    # Debug: Log the role value
+    print(f"🔍 DEBUG: ingest_document received role='{role}' (type: {type(role)})", file=sys.stderr)
+    
+    if not BACKEND_BASE_URL:
+        return "❌ Backend URL is not configured. Please set BACKEND_BASE_URL environment variable or ensure it defaults to http://localhost:8000"
+    
     if not tenant_id or not tenant_id.strip():
         return "❗ Tenant ID is required to ingest documents."
     
+    # Ensure role is not None or empty
+    if not role or not role.strip():
+        role = DEFAULT_ROLE
+        print(f"⚠️ WARNING: Role was empty/None in ingest_document, defaulting to '{role}'", file=sys.stderr)
+    role = role.strip()
+    
     if not can_ingest_documents(role):
         return "❌ Access Denied: You need Editor, Admin, or Owner role to ingest documents."
 
@@ -373,10 +388,14 @@ def ingest_document(
     }
 
     try:
+        # Ensure role is set correctly for the header
+        final_role = role.strip() if role and role.strip() else DEFAULT_ROLE
+        print(f"🔍 DEBUG: Sending request with role='{final_role}' in x-user-role header", file=sys.stderr)
+        
         headers = {
             "Content-Type": "application/json",
             "x-tenant-id": tenant_id,
-            "x-user-role": role if role else DEFAULT_ROLE
+            "x-user-role": final_role
         }
         response = requests.post(
             f"{BACKEND_BASE_URL}/rag/ingest-document",
@@ -413,6 +432,14 @@ def ingest_document(
                     message += f"- **Extraction Method:** {method}\n"
             
             return message
+        elif response.status_code == 403:
+            # Permission denied - show clear message
+            try:
+                error_data = response.json()
+                error_detail = error_data.get('detail', response.text)
+            except:
+                error_detail = response.text
+            return f"🔒 **Permission Denied (403):**\n\n{error_detail}\n\n**Solution:** Change your **User Role** dropdown (top right) from 'viewer' to 'editor', 'admin', or 'owner' and try again."
         return f"❌ Ingestion failed ({response.status_code}): {response.text}"
     except requests.exceptions.ConnectionError:
         return "❌ Could not reach the backend. Make sure the FastAPI server is running."
@@ -423,6 +450,9 @@ def ingest_document(
 
 
 def ingest_file(tenant_id: str, role: str, file_obj):
+    if not BACKEND_BASE_URL:
+        return "❌ Backend URL is not configured. Please set BACKEND_BASE_URL environment variable or ensure it defaults to http://localhost:8000"
+    
     if not tenant_id or not tenant_id.strip():
         return "❗ Tenant ID is required to ingest files."
     if file_obj is None:
@@ -1618,32 +1648,21 @@ with gr.Blocks(
         box-shadow: 0 18px 60px rgba(15, 23, 42, 1);
     }
 
-    .chatbot .message {
-        border-radius: 16px;
-        padding: 10px 14px;
-        font-size: 0.95rem;
-        line-height: 1.6;
-        max-width: 80%;
-    }
-
+    /* Keep Gradio's default layout, only adjust colors lightly */
     .chatbot .message.user {
-        margin-left: auto;
         background: #0ea5e9;
         color: #0b1020;
-        box-shadow: 0 12px 32px rgba(15, 23, 42, 0.9);
     }
 
     .chatbot .message.bot {
-        margin-right: auto;
         background: #020617;
-        border: 1px solid rgba(148, 163, 184, 0.8);
+        border-color: rgba(148, 163, 184, 0.8);
         color: #e5e7eb;
-        box-shadow: 0 14px 40px rgba(15, 23, 42, 1);
     }
 
     .chatbot .message.error {
-        background: linear-gradient(135deg, rgba(239, 68, 68, 0.16) 0%, rgba(127, 29, 29, 0.9) 100%);
-        border: 1px solid rgba(248, 113, 113, 0.9);
+        background: rgba(239, 68, 68, 0.18);
+        border-color: rgba(248, 113, 113, 0.9);
     }
     """
 ) as demo:
@@ -1934,10 +1953,18 @@ with gr.Blocks(
                 doc_id_value,
                 metadata
             ):
+                # Debug: Log the role value received
+                print(f"🔍 DEBUG: handle_ingest_document received role='{role}' (type: {type(role)})", file=sys.stderr)
+                
+                # Ensure role is not None or empty
+                if not role or role.strip() == "":
+                    role = DEFAULT_ROLE
+                    print(f"⚠️ WARNING: Role was empty/None, defaulting to '{role}'", file=sys.stderr)
+                
                 source_type = "raw_text" if mode == "Raw Text" else "url"
                 result = ingest_document(
                     tenant_id=tenant_id,
-                    role=role,
+                    role=role.strip() if role else DEFAULT_ROLE,
                     source_type=source_type,
                     content=content,
                     document_url=doc_url,

backend/api/mcp_clients/rag_client.py

@@ -64,11 +64,19 @@ class RAGClient:
         content: str,
         tenant_id: str,
         metadata: Optional[Dict[str, Any]] = None,
-        doc_id: Optional[str] = None
+        doc_id: Optional[str] = None,
+        user_role: Optional[str] = None
     ):
         """
         Sends content to the RAG server for ingestion with metadata.
         Returns the unwrapped data from the MCP server response.
+        
+        Args:
+            content: Text content to ingest
+            tenant_id: Tenant identifier
+            metadata: Optional metadata dictionary
+            doc_id: Optional document ID
+            user_role: User role (viewer, editor, admin, owner) - required for permission checks
         """
 
         try:
@@ -78,6 +86,10 @@ class RAGClient:
                     "content": content
                 }
                 
+                # Add role to payload (MCP server expects it for permission checks)
+                if user_role:
+                    payload["user_role"] = user_role
+                
                 # Add metadata if provided
                 if metadata:
                     payload["metadata"] = metadata
@@ -90,7 +102,14 @@ class RAGClient:
                 )
 
             if response.status_code != 200:
-                return {"error": f"HTTP {response.status_code}"}
+                error_text = response.text[:500] if hasattr(response, 'text') else f"HTTP {response.status_code}"
+                raise RuntimeError(
+                    f"RAG server returned error {response.status_code}: {error_text}\n\n"
+                    f"Please check:\n"
+                    f"1. RAG MCP server is running at {self.base_url}\n"
+                    f"2. Database connection (POSTGRESQL_URL) is configured\n"
+                    f"3. The 'documents' table exists in the database"
+                )
 
             data = response.json()
             
@@ -106,9 +125,23 @@ class RAGClient:
             # If not wrapped, return as-is (backward compatibility)
             return data
 
+        except httpx.RequestError as e:
+            error_msg = f"Failed to connect to RAG server at {self.base_url}: {str(e)}"
+            print(f"❌ RAG Ingest Connection Error: {error_msg}")
+            raise RuntimeError(
+                f"{error_msg}\n\n"
+                f"Please check:\n"
+                f"1. RAG_MCP_URL is set correctly (current: {self.base_url})\n"
+                f"2. RAG MCP server is running\n"
+                f"3. Network connectivity to the server"
+            ) from e
         except Exception as e:
-            print("RAG Ingest Error:", e)
-            return {"error": str(e)}
+            error_msg = f"RAG ingestion error: {str(e)}"
+            print(f"❌ {error_msg}")
+            raise RuntimeError(
+                f"{error_msg}\n\n"
+                f"Please check the RAG server logs for more details."
+            ) from e
 
     async def list_documents(self, tenant_id: str, limit: int = 1000, offset: int = 0):
         """

backend/api/routes/agent.py

@@ -148,9 +148,59 @@ Response:"""
             
             # STEP 2: ONLY IF NO RULES MATCHED - Proceed with normal flow
             yield f"data: {json.dumps({'status': 'classifying', 'message': 'Understanding your question...'})}\n\n"
+            
+            # Check if this is an admin identity question - handle it specially
+            user_text = agent_req.message.lower().strip()
+            user_text_normalized = " ".join(user_text.split())
+            admin_phrases = [
+                "who is the admin",
+                "who's the admin",
+                "who is admin",
+                "who is the administrator",
+                "who administers this platform",
+                "who is the owner",
+                "who owns this platform",
+                "who is the admin of integrachat",
+                "who administers integrachat",
+            ]
+            is_admin_question = (
+                any(p in user_text_normalized for p in admin_phrases) or
+                ("who" in user_text and "admin" in user_text)
+            )
+            
+            # For admin questions, ALWAYS check RAG first and answer directly from knowledge base
+            if is_admin_question:
+                yield f"data: {json.dumps({'status': 'searching', 'message': 'Searching knowledge base for admin information...'})}\n\n"
+                try:
+                    rag_prefetch = await orchestrator.mcp.call_rag(agent_req.tenant_id, agent_req.message)
+                    rag_results = []
+                    if isinstance(rag_prefetch, dict):
+                        rag_results = rag_prefetch.get("results") or rag_prefetch.get("hits") or []
+                    
+                    # If we have RAG hits, return the answer directly from the knowledge base
+                    if rag_results:
+                        best_hit = rag_results[0]
+                        admin_text = best_hit.get("text") or best_hit.get("content") or str(best_hit)
+                        response_text = f"According to the tenant knowledge base, {admin_text.strip()}"
+                    else:
+                        response_text = "I don't know who administers this platform based on the tenant data."
+                    
+                    # Stream the response word by word
+                    yield f"data: {json.dumps({'status': 'streaming', 'message': ''})}\n\n"
+                    import asyncio
+                    words = response_text.split()
+                    for word in words:
+                        yield f"data: {json.dumps({'token': word + ' ', 'done': False})}\n\n"
+                        await asyncio.sleep(0)
+                    yield f"data: {json.dumps({'token': '', 'done': True})}\n\n"
+                    return
+                except Exception as rag_err:
+                    # If RAG fails, fall through to normal flow
+                    pass
+            
             intent = await orchestrator.intent.classify(agent_req.message)
             
-            # Pre-fetch RAG if needed
+            # Pre-fetch RAG if needed (for non-admin questions)
             rag_results = []
             if intent == "rag" or "rag" in intent.lower():
                 yield f"data: {json.dumps({'status': 'searching', 'message': 'Searching knowledge base...'})}\n\n"
@@ -161,6 +211,16 @@ Response:"""
                 except Exception:
                     pass
             
+            # Also check if we have prefetched RAG results from earlier (for all questions)
+            # This ensures RAG context is used even if intent isn't "rag"
+            if not rag_results:
+                try:
+                    rag_prefetch = await orchestrator.mcp.call_rag(agent_req.tenant_id, agent_req.message)
+                    if isinstance(rag_prefetch, dict):
+                        rag_results = rag_prefetch.get("results") or rag_prefetch.get("hits") or []
+                except Exception:
+                    pass
+            
             # Build prompt with context
             if rag_results:
                 context = "\n\n".join([r.get("text", "")[:500] for r in rag_results[:3]])

backend/api/routes/rag.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter, Header, HTTPException, UploadFile, File, Form
+from fastapi import APIRouter, Header, HTTPException, UploadFile, File, Form, Request
 from pydantic import BaseModel
 from typing import Optional, Dict, Any
 from api.mcp_clients.rag_client import RAGClient
@@ -85,6 +85,7 @@ async def rag_ingest(
 @router.post("/ingest-document")
 async def rag_ingest_document(
     req: DocumentIngestRequest,
+    request: Request,
     x_tenant_id: Optional[str] = Header(None),
     x_user_role: str = Header("viewer")
 ):
@@ -114,26 +115,107 @@ async def rag_ingest_document(
     tenant_id = req.tenant_id or x_tenant_id
     if not tenant_id:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    
+    import sys
+    # Debug: Check actual headers received
+    all_headers = dict(request.headers)
+    print(f"🔍 DEBUG: All headers received: {list(all_headers.keys())}", file=sys.stderr)
+    print(f"🔍 DEBUG: x-user-role header value: '{all_headers.get('x-user-role', 'NOT FOUND')}'", file=sys.stderr)
+    print(f"🔍 DEBUG: x-user-role header value (case-insensitive): '{all_headers.get('X-User-Role', all_headers.get('x-user-role', 'NOT FOUND'))}'", file=sys.stderr)
+    print(f"🔍 DEBUG: Backend received x_user_role parameter='{x_user_role}' (type: {type(x_user_role)})", file=sys.stderr)
+    print(f"🔍 DEBUG: x_tenant_id header='{x_tenant_id}'", file=sys.stderr)
+    
     require_api_permission(x_user_role, "ingest_documents")
 
+    content_length = len(req.content) if req.content else 0
+    print(f"📥 Ingestion request received: tenant_id={tenant_id}, source_type={req.source_type}, content_length={content_length}", file=sys.stderr)
+    
+    # Validate content is not too short
+    if not req.content or not req.content.strip():
+        raise HTTPException(status_code=400, detail="Content cannot be empty. Please provide text to ingest.")
+    
+    if content_length < 10:
+        print(f"⚠️ Warning: Content is very short ({content_length} chars). This may result in no chunks being created.", file=sys.stderr)
+
     try:
+        print("🔧 Step 1: Preparing ingestion payload...", file=sys.stderr)
         # Prepare ingestion payload (async for URL fetching)
-        payload = await prepare_ingestion_payload(
-            tenant_id=tenant_id,
-            content=req.content,
-            source_type=req.source_type,
-            filename=req.metadata.get("filename") if req.metadata else None,
-            url=req.metadata.get("url") if req.metadata else None,
-            doc_id=req.metadata.get("doc_id") if req.metadata else None,
-            metadata=req.metadata
-        )
+        try:
+            payload = await prepare_ingestion_payload(
+                tenant_id=tenant_id,
+                content=req.content,
+                source_type=req.source_type,
+                filename=req.metadata.get("filename") if req.metadata else None,
+                url=req.metadata.get("url") if req.metadata else None,
+                doc_id=req.metadata.get("doc_id") if req.metadata else None,
+                metadata=req.metadata
+            )
+            print(f"✅ Step 1 complete: payload prepared", file=sys.stderr)
+        except Exception as prep_err:
+            print(f"❌ Step 1 FAILED (prepare_ingestion_payload): {prep_err}", file=sys.stderr)
+            import traceback
+            print(traceback.format_exc(), file=sys.stderr)
+            raise
         
+        print("🔧 Step 2: Processing ingestion with RAG client...", file=sys.stderr)
         # Process ingestion with metadata extraction
         extract_metadata = req.metadata.get("extract_metadata", True) if req.metadata else True
-        result = await process_ingestion(payload, rag_client, extract_metadata=extract_metadata)
+        try:
+            result = await process_ingestion(payload, rag_client, extract_metadata=extract_metadata, user_role=x_user_role)
+            print(f"✅ Step 2 complete: chunks_stored={result.get('chunks_stored', 0) if isinstance(result, dict) else 'N/A'}", file=sys.stderr)
+        except HTTPException:
+            # Re-raise HTTP exceptions (like 403 permission errors) as-is
+            raise
+        except Exception as proc_err:
+            # Check if it's a permission error with status_code attribute
+            if hasattr(proc_err, 'status_code') and proc_err.status_code == 403:
+                raise HTTPException(status_code=403, detail=getattr(proc_err, 'detail', str(proc_err)))
+            
+            print(f"❌ Step 2 FAILED (process_ingestion): {proc_err}", file=sys.stderr)
+            import traceback
+            print(traceback.format_exc(), file=sys.stderr)
+            raise
+        
+        # Check if ingestion actually succeeded
+        # First check if the result itself indicates an error
+        if isinstance(result, dict) and result.get('status') == 'error':
+            error_msg = result.get('message') or result.get('error') or "Unknown error from RAG server"
+            error_type = result.get('error_type', 'unknown')
+            print(f"❌ RAG server returned error ({error_type}): {error_msg}", file=sys.stderr)
+            
+            # If it's a permission error, return 403
+            if 'permission' in error_msg.lower() or 'not permitted' in error_msg.lower() or error_type == 'validation_error':
+                raise HTTPException(
+                    status_code=403,
+                    detail=f"Permission denied: {error_msg}\n\nPlease change your role to 'editor', 'admin', or 'owner' in the User Role dropdown."
+                )
+            else:
+                raise HTTPException(status_code=500, detail=f"RAG server error: {error_msg}")
+        
+        chunks_stored = result.get('chunks_stored', 0)
+        print(f"🔍 Debug: result keys={list(result.keys()) if isinstance(result, dict) else 'not a dict'}, chunks_stored={chunks_stored}", file=sys.stderr)
+        
+        if chunks_stored == 0:
+            # Get more details about why no chunks were stored
+            error_detail = result.get('error') or result.get('warnings') or result.get('message') or "No chunks were stored"
+            warnings = result.get('warnings')
+            
+            error_msg = f"Ingestion failed: {error_detail}"
+            if warnings:
+                error_msg += f"\nWarnings: {warnings}"
+            error_msg += (
+                "\n\nPossible causes:\n"
+                "1. Content too short or empty (minimum text required)\n"
+                "2. Database connection issue (check POSTGRESQL_URL in RAG server)\n"
+                "3. RAG MCP server error (check RAG server logs)\n"
+                "4. Database table 'documents' doesn't exist"
+            )
+            
+            print(f"❌ No chunks stored. Error detail: {error_detail}", file=sys.stderr)
+            raise HTTPException(status_code=500, detail=error_msg)
         
         # Build response message
-        message = f"Document ingested successfully. {result.get('chunks_stored', 0)} chunk(s) stored."
+        message = f"Document ingested successfully. {chunks_stored} chunk(s) stored."
         if result.get("extracted_metadata"):
             metadata_info = result["extracted_metadata"]
             if metadata_info.get("title"):
@@ -146,10 +228,55 @@ async def rag_ingest_document(
             "message": message,
             **result
         }
+    except HTTPException:
+        # Re-raise HTTP exceptions as-is
+        raise
     except ValueError as e:
-        raise HTTPException(status_code=400, detail=str(e))
+        import traceback
+        print(f"❌ Ingestion ValueError: {e}")
+        print(traceback.format_exc())
+        raise HTTPException(status_code=400, detail=f"Validation error: {str(e)}")
     except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
+        import traceback
+        import sys
+        error_detail = str(e)
+        error_type = type(e).__name__
+        full_traceback = traceback.format_exc()
+        
+        # Log to console with full details (use both stderr and stdout to ensure visibility)
+        error_log = f"❌ Ingestion Error ({error_type}): {error_detail}\nFull traceback:\n{full_traceback}"
+        print(error_log, file=sys.stderr)
+        print(error_log)  # Also print to stdout for uvicorn logs
+        
+        # Provide helpful error message
+        if "POSTGRESQL_URL" in error_detail or "database" in error_detail.lower() or "connection" in error_detail.lower():
+            error_msg = (
+                f"Database connection error: {error_detail}\n\n"
+                f"Please check:\n"
+                f"1. POSTGRESQL_URL is set correctly in your .env file\n"
+                f"2. Database is accessible\n"
+                f"3. The 'documents' table exists (run initialize_database() if needed)"
+            )
+        elif "RAG" in error_detail or "rag" in error_detail.lower() or "mcp" in error_detail.lower():
+            error_msg = (
+                f"RAG server error: {error_detail}\n\n"
+                f"Please check:\n"
+                f"1. RAG_MCP_URL is set correctly (default: http://localhost:8001)\n"
+                f"2. RAG MCP server is running\n"
+                f"3. Database connection (POSTGRESQL_URL) is configured in the RAG server"
+            )
+        else:
+            # For unknown errors, include the full error message
+            error_msg = f"Ingestion failed ({error_type}): {error_detail}"
+            # If it's a long traceback, include just the first few lines
+            if len(error_detail) > 500:
+                error_msg = f"Ingestion failed ({error_type}): {error_detail[:500]}...\n\nSee server logs for full traceback."
+        
+        # Ensure error message is not too long for HTTP response
+        if len(error_msg) > 2000:
+            error_msg = error_msg[:2000] + "...\n\n(Error message truncated. See server logs for full details.)"
+        
+        raise HTTPException(status_code=500, detail=error_msg)
 
 
 @router.post("/ingest-file")

backend/api/services/agent_orchestrator.py

@@ -610,12 +610,157 @@ Response:"""
                     return AgentResponse(text=json.dumps(admin_resp), decision=decision, tool_traces=tool_traces, reasoning_trace=reasoning_trace)
 
                 if decision.tool == "llm":
+                    # If the user is asking who the admin / owner is, try to ground the
+                    # answer in tenant-specific RAG before falling back to a generic LLM reply.
+                    user_text = req.message.lower()
+                    # Normalize whitespace to make matching more robust
+                    user_text_normalized = " ".join(user_text.split())
+                    admin_phrases = [
+                        "who is the admin",
+                        "who's the admin",
+                        "who is admin",
+                        "who is the administrator",
+                        "who's the administrator",
+                        "who administers this platform",
+                        "who administers the platform",
+                        "who is the owner",
+                        "who's the owner",
+                        "who owns this platform",
+                        "who owns the platform",
+                        "who is the admin of integrachat",
+                        "who's the admin of integrachat",
+                    ]
+                    use_rag_for_admin = any(p in user_text_normalized for p in admin_phrases) or (
+                        "admin" in user_text and "who" in user_text
+                    )
+
+                    prompt_for_llm = req.message
+
+                    if use_rag_for_admin:
+                        try:
+                            rag_start = time.time()
+                            rag_resp = await self.rag_with_repair(
+                                query=req.message,
+                                tenant_id=req.tenant_id,
+                                original_threshold=0.2,
+                                reasoning_trace=reasoning_trace,
+                                user_id=req.user_id,
+                            )
+                            rag_latency_ms = int((time.time() - rag_start) * 1000)
+                            tools_used.append("rag")
+
+                            rag_formatted = self._format_tool_output("rag", rag_resp, rag_latency_ms)
+                            tool_traces.append({"tool": "rag", "response": rag_formatted})
+
+                            hits = self._extract_hits(rag_formatted)
+                            hits_count = len(hits)
+                            avg_score = rag_formatted.get("avg_score")
+                            top_score = rag_formatted.get("top_score")
+
+                            self._analytics_log_tool_usage(
+                                tenant_id=req.tenant_id,
+                                tool_name="rag",
+                                latency_ms=rag_latency_ms,
+                                success=True,
+                                user_id=req.user_id,
+                            )
+
+                            reasoning_trace.append(
+                                {
+                                    "step": "tool_execution",
+                                    "tool": "rag",
+                                    "hit_count": hits_count,
+                                    "top_score": top_score,
+                                    "avg_score": avg_score,
+                                    "summary": self._summarize_hits(rag_formatted, limit=2),
+                                    "note": "admin_identity_override",
+                                }
+                            )
+
+                            # For admin questions, answer directly from RAG and avoid any
+                            # generic LLM behaviour. If there is at least one hit, return
+                            # that snippet; otherwise return an explicit "don't know".
+                            if hits:
+                                best = hits[0]
+                                admin_text = best.get("text") or best.get("content") or str(best)
+                                llm_out = f"According to the tenant knowledge base, {admin_text.strip()}"
+                            else:
+                                llm_out = "I don't know who administers this platform based on the tenant data."
+
+                            llm_latency_ms = 0
+                            estimated_tokens = len(llm_out) // 4 + len(req.message) // 4
+                            total_tokens += estimated_tokens
+
+                            self._analytics_log_tool_usage(
+                                tenant_id=req.tenant_id,
+                                tool_name="llm",
+                                latency_ms=llm_latency_ms,
+                                tokens_used=estimated_tokens,
+                                success=True,
+                                user_id=req.user_id,
+                            )
+
+                            reasoning_trace.append(
+                                {
+                                    "step": "llm_response",
+                                    "mode": "admin_from_rag_only",
+                                    "latency_ms": llm_latency_ms,
+                                    "estimated_tokens": estimated_tokens,
+                                }
+                            )
+
+                            total_latency_ms = int((time.time() - start_time) * 1000)
+                            self._analytics_log_agent_query(
+                                tenant_id=req.tenant_id,
+                                message_preview=req.message[:200],
+                                intent=intent,
+                                tools_used=tools_used,
+                                total_tokens=total_tokens,
+                                total_latency_ms=total_latency_ms,
+                                success=True,
+                                user_id=req.user_id,
+                            )
+
+                            return AgentResponse(text=llm_out, decision=decision, reasoning_trace=reasoning_trace)
+
+                        except Exception as rag_err:
+                            reasoning_trace.append(
+                                {
+                                    "step": "rag_for_admin_fallback",
+                                    "status": "error",
+                                    "error": str(rag_err),
+                                }
+                            )
+
+                    # For all other questions, if we already have RAG hits from pgvector
+                    # (rag_results from the prefetch step), reuse them to ground the
+                    # LLM response instead of answering purely from the model.
+                    if not use_rag_for_admin and rag_results:
+                        try:
+                            rag_prefetched_dict: Dict[str, Any] = {"results": rag_results}
+                            prompt_for_llm = self._build_prompt_with_rag(req, rag_prefetched_dict)
+                            reasoning_trace.append(
+                                {
+                                    "step": "rag_context_for_llm",
+                                    "hit_count": len(rag_results),
+                                    "note": "used_prefetched_pgvector_hits",
+                                }
+                            )
+                        except Exception as build_err:
+                            reasoning_trace.append(
+                                {
+                                    "step": "rag_context_for_llm",
+                                    "status": "error",
+                                    "error": str(build_err),
+                                }
+                            )
+
                     llm_start = time.time()
-                    llm_out = await self.llm.simple_call(req.message, temperature=req.temperature)
+                    llm_out = await self.llm.simple_call(prompt_for_llm, temperature=req.temperature)
                     llm_latency_ms = int((time.time() - llm_start) * 1000)
                     tools_used.append("llm")
                     
-                    estimated_tokens = len(llm_out) // 4 + len(req.message) // 4
+                    estimated_tokens = len(llm_out) // 4 + len(prompt_for_llm) // 4
                     total_tokens += estimated_tokens
                     
                     self._analytics_log_tool_usage(
@@ -1046,7 +1191,73 @@ Response:"""
             # Build comprehensive prompt with all collected data
             data_section = "\n---\n".join(collected_data) if collected_data else ""
 
-        # Build final prompt
+        # Build final response. For admin-identity style questions, bypass generic
+        # multi-step LLM behaviour and answer directly from RAG data if available.
+        user_text = req.message.lower()
+        user_text_normalized = " ".join(user_text.split())
+        admin_phrases = [
+            "who is the admin",
+            "who's the admin",
+            "who is admin",
+            "who is the administrator",
+            "who's the administrator",
+            "who administers this platform",
+            "who administers the platform",
+            "who is the owner",
+            "who's the owner",
+            "who owns this platform",
+            "who owns the platform",
+            "who is the admin of integrachat",
+            "who's the admin of integrachat",
+        ]
+        if any(p in user_text_normalized for p in admin_phrases) or ("admin" in user_text and "who" in user_text):
+            hits = self._extract_hits(rag_data) if rag_data else []
+            if hits:
+                best = hits[0]
+                admin_text = best.get("text") or best.get("content") or str(best)
+                llm_out = f"According to the tenant knowledge base, {admin_text.strip()}"
+            else:
+                llm_out = "I don't know who administers this platform based on the tenant data."
+
+            llm_latency_ms = 0
+            estimated_tokens = len(llm_out) // 4 + len(req.message) // 4
+            total_tokens += estimated_tokens
+            tools_used.append("llm")
+
+            self._analytics_log_tool_usage(
+                tenant_id=req.tenant_id,
+                tool_name="llm",
+                latency_ms=llm_latency_ms,
+                tokens_used=estimated_tokens,
+                success=True,
+                user_id=req.user_id
+            )
+
+            total_latency_ms = int((time.time() - start_time) * 1000)
+            self._analytics_log_agent_query(
+                tenant_id=req.tenant_id,
+                message_preview=req.message[:200],
+                intent="multi_step",
+                tools_used=tools_used,
+                total_tokens=total_tokens,
+                total_latency_ms=total_latency_ms,
+                success=True,
+                user_id=req.user_id
+            )
+
+            return AgentResponse(
+                text=llm_out,
+                decision=decision,
+                tool_traces=tool_traces,
+                reasoning_trace=reasoning_trace + [{
+                    "step": "llm_response",
+                    "mode": "multi_step_admin_from_rag_only",
+                    "latency_ms": llm_latency_ms,
+                    "estimated_tokens": estimated_tokens
+                }]
+            )
+
+        # Otherwise, build the normal multi-step synthesis prompt.
         if data_section:
             prompt = (
                 f"You are an assistant helping tenant {req.tenant_id}.\n\n"
@@ -1061,7 +1272,6 @@ Response:"""
                 f"and practical steps whenever possible. If the information is incomplete, explain "
                 f"what can and cannot be concluded from the available data."
             )
-
         else:
             # No data collected, just answer the question
             prompt = req.message
@@ -1072,10 +1282,10 @@ Response:"""
             llm_out = await self.llm.simple_call(prompt, temperature=req.temperature)
             llm_latency_ms = int((time.time() - llm_start) * 1000)
             tools_used.append("llm")
-            
+
             estimated_tokens = len(llm_out) // 4 + len(prompt) // 4
             total_tokens += estimated_tokens
-            
+
             self._analytics_log_tool_usage(
                 tenant_id=req.tenant_id,
                 tool_name="llm",
@@ -1084,7 +1294,7 @@ Response:"""
                 success=True,
                 user_id=req.user_id
             )
-            
+
             total_latency_ms = int((time.time() - start_time) * 1000)
             self._analytics_log_agent_query(
                 tenant_id=req.tenant_id,
@@ -1096,7 +1306,7 @@ Response:"""
                 success=True,
                 user_id=req.user_id
             )
-            
+
             return AgentResponse(
                 text=llm_out,
                 decision=decision,

backend/api/services/document_ingestion.py

@@ -217,7 +217,8 @@ async def prepare_ingestion_payload(
 async def process_ingestion(
     payload: Dict[str, Any],
     rag_client,
-    extract_metadata: bool = True
+    extract_metadata: bool = True,
+    user_role: Optional[str] = None
 ) -> Dict[str, Any]:
     """
     Process the ingestion payload by sending it to the RAG MCP server.
@@ -260,22 +261,71 @@ async def process_ingestion(
     }
     
     # Send to RAG MCP server with metadata
-    result = await rag_client.ingest_with_metadata(
-        content=content,
-        tenant_id=tenant_id,
-        metadata=final_metadata,
-        doc_id=doc_id
-    )
-    
-    # Enhance result with metadata
-    return {
-        "status": "ok",
-        "tenant_id": tenant_id,
-        "source_type": source_type,
-        "doc_id": doc_id,
-        "chunks_stored": result.get("chunks_stored", 0),
-        "metadata": final_metadata,
-        "extracted_metadata": extracted_metadata,  # Include extracted metadata in response
-        **result
-    }
+    try:
+        result = await rag_client.ingest_with_metadata(
+            content=content,
+            tenant_id=tenant_id,
+            metadata=final_metadata,
+            doc_id=doc_id,
+            user_role=user_role
+        )
+        
+        # Check if result indicates an error (multiple ways the RAG server can signal errors)
+        if isinstance(result, dict):
+            # Check for explicit error status
+            if result.get("status") == "error":
+                error_msg = result.get("message") or result.get("error") or "Unknown error from RAG server"
+                error_type = result.get("error_type", "unknown_error")
+                logger.error(f"RAG ingestion error ({error_type}): {error_msg}")
+                
+                # For permission errors, raise a specific exception that can be caught and converted to HTTPException
+                if "permission" in error_msg.lower() or "not permitted" in error_msg.lower() or error_type == "validation_error":
+                    # Create a custom exception that will be caught and converted to HTTPException
+                    class PermissionError(Exception):
+                        pass
+                    perm_err = PermissionError(f"Permission denied: {error_msg}")
+                    perm_err.status_code = 403
+                    perm_err.detail = f"Permission denied: {error_msg}\n\nPlease change your role to 'editor', 'admin', or 'owner' in the User Role dropdown in app.py."
+                    raise perm_err
+                
+                raise ValueError(f"RAG server error ({error_type}): {error_msg}")
+            
+            # Check for error field
+            if "error" in result:
+                error_msg = result.get("error", "Unknown error from RAG server")
+                logger.error(f"RAG ingestion error: {error_msg}")
+                raise ValueError(f"RAG server error: {error_msg}")
+        
+        chunks_stored = result.get("chunks_stored", 0) if isinstance(result, dict) else 0
+        
+        # Enhance result with metadata
+        response = {
+            "status": "ok",
+            "tenant_id": tenant_id,
+            "source_type": source_type,
+            "doc_id": doc_id,
+            "chunks_stored": chunks_stored,
+            "metadata": final_metadata,
+            "extracted_metadata": extracted_metadata,  # Include extracted metadata in response
+        }
+        
+        # Add any additional fields from result if it's a dict
+        if isinstance(result, dict):
+            response.update(result)
+        
+        return response
+    except Exception as e:
+        # Re-raise permission errors as-is (they'll be caught and converted to HTTPException)
+        if hasattr(e, 'status_code') and e.status_code == 403:
+            raise
+        
+        logger.error(f"Failed to ingest document to RAG server: {e}", exc_info=True)
+        # Re-raise with more context
+        raise RuntimeError(
+            f"Failed to send document to RAG MCP server: {str(e)}\n\n"
+            f"Please check:\n"
+            f"1. RAG_MCP_URL is set correctly (default: http://localhost:8001)\n"
+            f"2. RAG MCP server is running\n"
+            f"3. Database connection (POSTGRESQL_URL) is configured in the RAG server"
+        ) from e
 

backend/mcp_server/common/database.py

@@ -137,11 +137,22 @@ def insert_document_chunks(tenant_id: str, text: str, embedding: list, metadata:
         metadata: Optional JSON metadata (title, summary, tags, topics, etc.)
         doc_id: Optional document ID to group chunks from the same document
     """
+    import json
+    import traceback
+    
+    # Normalize tenant_id to ensure consistency
+    tenant_id = tenant_id.strip()
+    
+    if not tenant_id:
+        raise ValueError("tenant_id cannot be empty")
+    
+    if not text or not text.strip():
+        raise ValueError("text cannot be empty")
+    
+    if not embedding or len(embedding) != 384:
+        raise ValueError(f"embedding must be a 384-dimensional vector, got {len(embedding) if embedding else 0} dimensions")
+    
     try:
-        import json
-        # Normalize tenant_id to ensure consistency
-        tenant_id = tenant_id.strip()
-        
         conn = get_connection()
         cur = conn.cursor()
 
@@ -159,10 +170,25 @@ def insert_document_chunks(tenant_id: str, text: str, embedding: list, metadata:
         conn.commit()
         cur.close()
         conn.close()
+        
+        print(f"✅ DB INSERT: Successfully inserted chunk for tenant '{tenant_id}' (doc_id: {doc_id or 'N/A'})")
 
-    except Exception as e:
-        print("DB INSERT ERROR:", e)
+    except ValueError as ve:
+        # Re-raise ValueError as-is (validation errors)
+        print(f"❌ DB INSERT VALIDATION ERROR: {ve}")
         raise
+    except Exception as e:
+        error_msg = f"DB INSERT ERROR (tenant_id='{tenant_id}'): {str(e)}"
+        print(f"❌ {error_msg}")
+        print(traceback.format_exc())
+        # Wrap in a more descriptive error
+        raise RuntimeError(
+            f"Failed to insert document into database: {str(e)}\n"
+            f"Please check:\n"
+            f"1. POSTGRESQL_URL is set correctly in .env\n"
+            f"2. Database is accessible and pgvector extension is installed\n"
+            f"3. Documents table exists (run initialize_database() if needed)"
+        ) from e
 
 
 def search_vectors(tenant_id: str, vector: list, limit: int = 5) -> List[Dict[str, Any]]:

backend/mcp_server/rag/ingest.py

@@ -45,22 +45,46 @@ async def rag_ingest(context: TenantContext, payload: Mapping[str, object]) -> d
         raise ToolValidationError("no text detected after preprocessing")
 
     stored = 0
-    for chunk in chunks:
-        vector = embed_text(chunk)
-        # Store metadata with each chunk (same metadata for all chunks from same document)
-        insert_document_chunks(
-            context.tenant_id,
-            chunk,
-            vector,
-            metadata=metadata,
-            doc_id=doc_id
+    errors = []
+    
+    for i, chunk in enumerate(chunks):
+        try:
+            vector = embed_text(chunk)
+            # Store metadata with each chunk (same metadata for all chunks from same document)
+            insert_document_chunks(
+                context.tenant_id,
+                chunk,
+                vector,
+                metadata=metadata,
+                doc_id=doc_id
+            )
+            stored += 1
+        except Exception as e:
+            error_msg = f"Failed to store chunk {i+1}/{len(chunks)}: {str(e)}"
+            errors.append(error_msg)
+            print(f"❌ {error_msg}")
+            # Continue with other chunks, but log the error
+    
+    if stored == 0:
+        # If no chunks were stored, raise an error
+        error_summary = "\n".join(errors) if errors else "Unknown error during database insertion"
+        raise ToolValidationError(
+            f"Failed to store any chunks to database. Errors:\n{error_summary}\n\n"
+            f"Please check:\n"
+            f"1. POSTGRESQL_URL is set correctly in your .env file\n"
+            f"2. Database is accessible and the 'documents' table exists\n"
+            f"3. pgvector extension is installed in your PostgreSQL database"
         )
-        stored += 1
+    
+    if errors:
+        # Some chunks failed, but some succeeded - return a warning
+        print(f"⚠️ WARNING: {len(errors)} chunk(s) failed to store, but {stored} chunk(s) were stored successfully")
 
     return {
         "tenant_id": context.tenant_id,
         "chunks_ingested": stored,
         "metadata": {"chunk_words": max_words_value, **(metadata or {})},
         "doc_id": doc_id,
+        "warnings": errors if errors else None,
     }