Upload 465 files

SuperAgent/app/__init__.py

@@ -0,0 +1 @@
+# Package marker

SuperAgent/app/models/schemas.py

@@ -0,0 +1,24 @@
+from typing import Any, Literal, Optional
+from pydantic import BaseModel, Field
+
+# Reuse the basic structure for compatibility
+class ChatRequest(BaseModel):
+    message: str
+    asset_id: Optional[str] = None
+    session_id: Optional[str] = "default"
+    user_id: Optional[str] = None  # Resolved from "User_ID: question" prefix or DEFAULT_USER_ID env var
+
+
+class ChatResponse(BaseModel):
+    answer: str
+    intent: Literal["live", "analysis", "anomaly", "report"]
+    analysis: Optional[Any] = None # Will hold AnalysisResponse if intent is analysis
+    data: Optional[Any] = None # Will hold rows if intent is live, or complex reports if anomaly
+    generated_sql: Optional[str] = None
+    orchestration: Optional[list[str]] = None
+
+class SessionCreateRequest(BaseModel):
+    user_id: Optional[str] = None
+
+class SessionTitleUpdateRequest(BaseModel):
+    title: str

SuperAgent/app/services/asset_registry.py

@@ -0,0 +1,390 @@
+import os
+import re
+import time
+import logging
+from typing import Dict, List, Optional, Any
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+
+logger = logging.getLogger(__name__)
+
+class AssetRegistry:
+    """
+    Centralized Asset Registry for the SuperAgent.
+    Standardizes asset naming and mapping across all agents.
+    """
+    def __init__(self):
+        db_url = os.environ.get('DATABASE_URL', 'postgresql://postgres:postgres@109.176.197.144:55432/openremote')
+        if db_url.startswith("postgresql://"):
+            db_url = db_url.replace("postgresql://", "postgresql+psycopg://", 1)
+        
+        self.schema = os.environ.get('OPENREMOTE_SCHEMA', 'openremote')
+        self.engine = create_engine(db_url, pool_pre_ping=True)
+        self.SessionLocal = sessionmaker(bind=self.engine)
+        
+        self._cache: List[Dict[str, Any]] = []
+        self._mappings: Dict[str, Dict[str, Any]] = {} # id -> asset
+        self._name_map: Dict[str, str] = {} # normalized_name -> id
+        self._last_load = 0
+        self._ttl = 300 # 5 minutes
+        self._user_cache: Dict[str, List[Dict[str, Any]]] = {} # user_id -> allowed_assets
+        self._user_last_load: Dict[str, float] = {}
+
+    def _load_user_assets(self, user_id: str):
+        now = time.time()
+        if user_id in self._user_cache and (now - self._user_last_load.get(user_id, 0) < self._ttl):
+            return
+
+        query = text(f"""
+            WITH RECURSIVE hierarchy AS (
+                SELECT a.id::text, a.name, a.parent_id::text, a.attributes::jsonb AS attributes
+                FROM {self.schema}.asset a
+                JOIN {self.schema}.user_asset_link ual ON a.id = ual.asset_id
+                WHERE ual.user_id = :user_id
+
+                UNION
+
+                SELECT child.id::text, child.name, child.parent_id::text, child.attributes::jsonb AS attributes
+                FROM {self.schema}.asset child
+                JOIN hierarchy h ON child.parent_id = h.id
+            )
+            SELECT DISTINCT id, name, parent_id, attributes
+            FROM hierarchy
+            ORDER BY name
+        """)
+        
+        try:
+            with self.SessionLocal() as session:
+                result = session.execute(query, {"user_id": user_id})
+                assets = [dict(row) for row in result.mappings().all()]
+                
+                self._user_cache[user_id] = assets
+                self._user_last_load[user_id] = now
+                logger.info(f"Asset Registry: Loaded {len(assets)} assets for user {user_id}.")
+        except Exception as e:
+            logger.error(f"Asset Registry: Failed to load assets for user {user_id}: {e}")
+            self._user_cache[user_id] = []
+
+    # def normalize_name(self, name: str) -> str:
+    #     """Standardizes name for fuzzy matching."""
+    #     if not name: return ""
+    #     # Remove special characters and spaces, lowercase
+    #     return re.sub(r'[^a-zA-Z0-9]', '', name).lower()
+
+    def normalize_name(self, name: str) -> str:
+        """Standardizes name for fuzzy matching."""
+        if not name: return ""
+        # Remove special characters and spaces, lowercase
+        normalized = re.sub(r'[^a-zA-Z0-9]', '', name).lower()
+        # Remove leading zeros from numbers (e.g. coldroom01 -> coldroom1, but keeps 102 as 102)
+        normalized = re.sub(r'(?<!\d)0+(\d+)', r'\1', normalized)
+        return normalized
+
+
+    def get_summary(self, user_id: str) -> Dict[str, Any]:
+        """Returns a statistical summary of assets allowed for the user."""
+        self._load_user_assets(user_id)
+        assets = self._user_cache.get(user_id, [])
+        
+        stats = {
+            "Total": len(assets),
+            "Tanks": 0,
+            "ColdRooms": 0,
+            "Machines": 0,
+            "Boilers": 0,
+            "SmokeAlarms": 0,
+            "Other": 0
+        }
+        
+        for a in assets:
+            name = a['name'].lower()
+            if 'tank' in name or 'physical' in name: stats["Tanks"] += 1
+            elif 'cold' in name: stats["ColdRooms"] += 1
+            elif 'machine' in name or 'filling' in name: stats["Machines"] += 1
+            elif 'boiler' in name: stats["Boilers"] += 1
+            elif 'smoke' in name or 'alarm' in name: stats["SmokeAlarms"] += 1
+            else: stats["Other"] += 1
+            
+        return stats
+
+    def _is_general_status_query(self, query: str) -> bool:
+        normalized = "".join(query.lower().split())
+        for kw in ["systemstatus", "statusofall", "allassets", "alldevices", "completestatus", "statusreport"]:
+            if kw in normalized:
+                return True
+        return False
+
+    def is_prefix_number_code(self, name: str) -> bool:
+        normalized = re.sub(r'[^a-zA-Z0-9]', '', name).lower()
+        return bool(re.match(r'^[a-zA-Z]+\d+$', normalized))
+
+    def is_asset_mentioned(self, asset_name: str, query: str) -> bool:
+        """
+        Robust check if asset_name is mentioned in the query,
+        respecting word boundaries and avoiding substring overlap (e.g. "st1" inside "past10").
+        """
+        q = query.lower()
+        an = asset_name.lower()
+        
+        # 1. Exact or direct substring with word boundaries in original query
+        words = re.findall(r'[a-zA-Z]+|\d+', an)
+        if not words:
+            return False
+            
+        # Build a regex pattern that matches the words in order, allowing optional spaces/hyphens/underscores
+        pattern_parts = []
+        for i, w in enumerate(words):
+            pattern_parts.append(re.escape(w))
+            if i < len(words) - 1:
+                pattern_parts.append(r'\s*[-_]?\s*')
+                
+        pattern = r'\b' + "".join(pattern_parts) + r'\b'
+        
+        try:
+            if re.search(pattern, q):
+                return True
+        except Exception:
+            pass
+            
+        # 1b. Check prefix-stripped suffix matching (e.g., "ST1 Weather Station" matches when query has "weather station")
+        match = re.match(r'^([a-zA-Z]+\d+)\s+(.+)$', asset_name.strip())
+        if match:
+            suffix = match.group(2)
+            if len(suffix) > 3:
+                if self.is_asset_mentioned(suffix, query):
+                    return True
+            
+        # 2. Normalized fallback (only for names longer than 4 characters to avoid ST1/F1 false positives)
+        norm_q = self.normalize_name(q)
+        norm_an = self.normalize_name(an)
+        if len(norm_an) > 4 and not self.is_prefix_number_code(an) and norm_an in norm_q:
+            return True
+            
+        return False
+
+    def find_mentioned_assets(self, query_name: str, user_id: str) -> List[Dict[str, Any]]:
+        self._load_user_assets(user_id)
+        allowed_assets = self._user_cache.get(user_id, [])
+        
+        is_general_status = self._is_general_status_query(query_name)
+        mentioned = []
+        
+        for a in allowed_assets:
+            if is_general_status and a['name'].lower() == 'status':
+                continue
+            if self.is_asset_mentioned(a['name'], query_name):
+                mentioned.append(a)
+                
+        # If both a physical/non-Status asset and 'Status' are matched, discard 'Status'
+        if len(mentioned) >= 2:
+            has_physical = any(a['name'].lower() != 'status' for a in mentioned)
+            if has_physical:
+                mentioned = [a for a in mentioned if a['name'].lower() != 'status']
+
+        return mentioned
+
+    def find_asset(self, query_name: str, user_id: str) -> Optional[Dict[str, Any]]:
+        """Finds an asset by name using robust matching, filtered by user permissions."""
+        self._load_user_assets(user_id)
+        allowed_assets = self._user_cache.get(user_id, [])
+
+        # ---------------------------------------------------------------
+        # PRE-RESOLUTION: "Tank N" -> "physicalNN" (zero-padded, 2 digits)
+        # This mirrors the same rule in agents/live/system_message.md so
+        # that both the Live path and the Report path resolve to the same
+        # physical asset regardless of which path processes the message.
+        # ---------------------------------------------------------------
+        # tank_alias_match = re.search(r'\btank\s*(\d+)\b', query_name, re.IGNORECASE)
+        # if tank_alias_match:
+        #     tank_num = int(tank_alias_match.group(1))
+        #     physical_name = f'physical{tank_num:02d}'  # e.g. physical11
+        #     for a in allowed_assets:
+        #         if self.normalize_name(a['name']) == physical_name:
+        #             logger.debug(f"Asset Registry: 'Tank {tank_num}' pre-resolved to '{a['name']}' ({a['id']})")
+        #             return a
+
+                # 1. Multi-tank comparison check
+        tank_matches = re.findall(r'\btank\s*(\d+)\b', query_name, re.IGNORECASE)
+        if len(tank_matches) >= 2:
+            matched_assets = []
+            for t_num in tank_matches:
+                physical_name = f'physical{int(t_num):02d}'
+                for a in allowed_assets:
+                    if self.normalize_name(a['name']) == self.normalize_name(physical_name):
+                        matched_assets.append(a)
+                        break
+            # Multiple allowed tanks
+            if len(matched_assets) >= 2:
+                parent_ids = {a.get('parent_id') for a in matched_assets if a.get('parent_id')}
+                if len(parent_ids) == 1:
+                    parent_id = list(parent_ids)[0]
+                    parent_asset = self.get_asset_by_id(parent_id, user_id)
+                    if parent_asset:
+                        logger.info(f"Asset Registry: Multi-tank comparison detected. Mapping to common parent '{parent_asset['name']}' ({parent_asset['id']})")
+                        return parent_asset
+
+        # 2. General Multi-asset comparison check (For Coldrooms, Machines, etc.)
+        mentioned = self.find_mentioned_assets(query_name, user_id)
+        if len(mentioned) >= 2:
+            parent_ids = {a.get('parent_id') for a in mentioned if a.get('parent_id')}
+            if len(parent_ids) == 1:
+                parent_id = list(parent_ids)[0]
+                parent_asset = self.get_asset_by_id(parent_id, user_id)
+                if parent_asset:
+                    logger.info(f"Asset Registry: Multi-asset comparison detected. Mapping to common parent '{parent_asset['name']}' ({parent_asset['id']})")
+                    return parent_asset
+
+        # 2. Single tank fallback
+        tank_alias_match = re.search(r'\btank\s*(\d+)\b', query_name, re.IGNORECASE)
+        if tank_alias_match:
+            tank_num = int(tank_alias_match.group(1))
+            physical_name = f'physical{tank_num:02d}'  # e.g. physical11
+            for a in allowed_assets:
+                if self.normalize_name(a['name']) == self.normalize_name(physical_name):
+                    logger.debug(f"Asset Registry: 'Tank {tank_num}' pre-resolved to '{a['name']}' ({a['id']})")
+                    return a
+
+        
+        normalized_query = self.normalize_name(query_name)
+        if not normalized_query:
+            return None
+        
+        is_general_status = self._is_general_status_query(query_name)
+        is_tank = any(k in normalized_query for k in ['tank', 'physical', 'refinery']) or bool(re.search(r'\bst\s*\d+\b', query_name.lower())) or bool(re.search(r'\bst\d+\b', normalized_query))
+        is_cold = 'cold' in normalized_query
+        is_boiler = 'boiler' in normalized_query
+        is_machine = any(k in normalized_query for k in ['machine', 'filling'])
+
+        # Find target number in query (excluding time/limits)
+        num_matches = re.finditer(r'\b(\d+)\b', query_name)
+        target_num = None
+        for m in num_matches:
+            val = int(m.group(1))
+            start, end = m.span()
+            surrounding = query_name[max(0, start - 15):min(len(query_name), end + 15)].lower()
+            if re.search(
+                r'\b\d+\s*(?:minute|min|hour|hr|day|week|month|read(?:ing|ngs?|ings?)|row|log|point|datapoint)s?\b',
+                surrounding
+            ):
+                continue
+            target_num = val
+            break
+
+        if target_num is None:
+            num_match = re.search(r'(\d+)', normalized_query)
+            target_num = int(num_match.group(1)) if num_match else None
+
+        candidate_scores = []
+        for a in allowed_assets:
+            if is_general_status and a['name'].lower() == 'status':
+                continue
+
+            name = a['name']
+            db_name_lower = name.lower()
+            normalized_db_name = self.normalize_name(name)
+            if not normalized_db_name:
+                continue
+
+            score = 0.0
+
+            # Rule 1: Exact Match (1.0)
+            if normalized_db_name == normalized_query:
+                score = 1.0
+
+            # Rule 2: Category + Number Match (0.98)
+            elif target_num is not None:
+                db_num_match = re.search(r'(\d+)', db_name_lower)
+                if db_num_match and int(db_num_match.group(1)) == target_num:
+                    cleaned_db = re.sub(r'[\s_-]', '', db_name_lower)
+                    is_db_tank = 'tank' in db_name_lower or 'physical' in db_name_lower or 'refinery' in db_name_lower or bool(re.match(r'^st\d+$', cleaned_db))
+                    
+                    category_matched = False
+                    if is_tank and is_db_tank:
+                        category_matched = True
+                    elif is_cold and 'cold' in db_name_lower:
+                        category_matched = True
+                    elif is_boiler and 'boiler' in db_name_lower:
+                        category_matched = True
+                    elif is_machine and ('machine' in db_name_lower or 'filling' in db_name_lower):
+                        category_matched = True
+
+                    if category_matched:
+                        score = 0.98
+
+            # Rule 3: Explicit Mention (0.95)
+            if score == 0.0:
+                if self.is_asset_mentioned(name, query_name):
+                    cleaned_db = re.sub(r'[\s_-]', '', db_name_lower)
+                    is_db_tank = 'tank' in db_name_lower or 'physical' in db_name_lower or 'refinery' in db_name_lower or bool(re.match(r'^st\d+$', cleaned_db))
+                    
+                    category_conflict = False
+                    if is_tank and not is_db_tank:
+                        category_conflict = True
+                    elif is_cold and 'cold' not in db_name_lower:
+                        category_conflict = True
+                    elif is_boiler and 'boiler' not in db_name_lower:
+                        category_conflict = True
+                    elif is_machine and not any(k in db_name_lower for k in ['machine', 'filling']):
+                        category_conflict = True
+
+                    if not category_conflict:
+                        score = 0.95
+
+            # Rule 4: General Category Match or Suffix removed match (0.70)
+            if score == 0.0:
+                is_general_category = (is_cold and 'cold' in db_name_lower) or \
+                                      (is_tank and ('tank' in db_name_lower or 'physical' in db_name_lower)) or \
+                                      (is_boiler and 'boiler' in db_name_lower) or \
+                                      (is_machine and ('machine' in db_name_lower or 'filling' in db_name_lower))
+                if is_general_category:
+                    score = 0.70
+                else:
+                    for suffix in ['tank', 'asset', 'sensor', 'machine', 'boiler', 'coldroom']:
+                        if normalized_query.endswith(suffix):
+                            short = normalized_query[:-len(suffix)]
+                            if normalized_db_name == short:
+                                score = 0.70
+                                break
+
+            # Rule 5: Generic Substring Match (0.60)
+            if score == 0.0:
+                if len(normalized_query) > 3 and normalized_db_name not in ['oil', 'water', 'status', 'temp', 'level']:
+                    if normalized_query in normalized_db_name or normalized_db_name in normalized_query:
+                        score = 0.60
+
+            if score > 0.0:
+                candidate_scores.append((a, score))
+
+        if not candidate_scores:
+            return None
+
+        # Sort candidate scores descending by score value
+        candidate_scores.sort(key=lambda x: x[1], reverse=True)
+        best_match, best_score = candidate_scores[0]
+
+        logger.info(f"Asset Registry: Best candidate '{best_match['name']}' with score {best_score}")
+
+        if best_score < 0.95:
+            logger.info(f"Asset Registry: Best score {best_score} is below 95% threshold. Asking counter-question.")
+            return None
+
+        # Check for ties on the highest score
+        ties = [item for item in candidate_scores if item[1] == best_score]
+        if len(ties) > 1:
+            logger.info(f"Asset Registry: Tie detected between {len(ties)} assets at score {best_score}. Asking counter-question.")
+            return None
+
+        return best_match
+
+    def get_asset_by_id(self, asset_id: str, user_id: str) -> Optional[Dict[str, Any]]:
+        self._load_user_assets(user_id)
+        for a in self._user_cache.get(user_id, []):
+            if a['id'] == asset_id:
+                return a
+        return None
+
+    def list_all_assets(self, user_id: str) -> List[Dict[str, Any]]:
+        self._load_user_assets(user_id)
+        return self._user_cache.get(user_id, [])
+
+asset_registry = AssetRegistry()

SuperAgent/app/services/email_service.py

@@ -0,0 +1,76 @@
+import os
+import smtplib
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from email.mime.base import MIMEBase
+from email import encoders
+import logging
+
+logger = logging.getLogger(__name__)
+
+class EmailService:
+    def __init__(self):
+        self.smtp_host = os.getenv("SMTP_HOST")
+        self.smtp_port = os.getenv("SMTP_PORT")
+        self.smtp_user = os.getenv("SMTP_USER")
+        self.smtp_pass = os.getenv("SMTP_PASSWORD")
+        self.smtp_from = os.getenv("SMTP_FROM", "[EMAIL_ADDRESS]")
+
+    def send_report(self, recipient: str, subject: str, body_html: str, attachment_path: str = None) -> bool:
+        if not recipient:
+            logger.warning("Email recipient is empty. Skipping email delivery.")
+            return False
+
+        if not self.smtp_host or not self.smtp_user or not self.smtp_pass:
+            logger.warning("SMTP credentials are not configured in environment. Printing email to console (MOCK).")
+            print(f"\n================ MOCK EMAIL SENT ================")
+            print(f"To: {recipient}")
+            print(f"From: {self.smtp_from}")
+            print(f"Subject: {subject}")
+            print(f"Body: {body_html[:300]}...")
+            if attachment_path:
+                print(f"Attachment: {os.path.basename(attachment_path)}")
+            print(f"=================================================\n")
+            return True
+
+        try:
+            # Create message container
+            msg = MIMEMultipart()
+            msg['From'] = self.smtp_from
+            msg['To'] = recipient
+            msg['Subject'] = subject
+
+            # Record the MIME types of both parts - text/html
+            msg.attach(MIMEText(body_html, 'html'))
+
+            # Attach PDF if provided
+            if attachment_path and os.path.exists(attachment_path):
+                filename = os.path.basename(attachment_path)
+                with open(attachment_path, "rb") as attachment:
+                    part = MIMEBase("application", "octet-stream")
+                    part.set_payload(attachment.read())
+                    encoders.encode_base64(part)
+                    part.add_header(
+                        "Content-Disposition",
+                        f"attachment; filename={filename}",
+                    )
+                    msg.attach(part)
+
+            # Connect to SMTP server
+            port = int(self.smtp_port) if self.smtp_port else 587
+            server = smtplib.SMTP(self.smtp_host, port)
+            server.starttls()
+            server.login(self.smtp_user, self.smtp_pass)
+            
+            # Send email
+            server.sendmail(self.smtp_from, recipient, msg.as_string())
+            server.quit()
+            
+            logger.info(f"Email sent successfully to {recipient}")
+            return True
+
+        except Exception as e:
+            logger.error(f"Failed to send email to {recipient}: {str(e)}")
+            return False
+
+email_service = EmailService()

SuperAgent/app/services/report_generator.py

@@ -0,0 +1,13 @@
+import os
+import sys
+
+# Dynamically append root folder of the project to sys.path to enable smooth resolution of agents package
+root_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
+if root_dir not in sys.path:
+    sys.path.insert(0, root_dir)
+
+# Import the core logic from the newly decoupled report agent service
+from agents.report.app.services.report_generator import report_generator_service, ReportGeneratorService
+
+# Maintain absolute backward compatibility for any consumers
+__all__ = ["report_generator_service", "ReportGeneratorService"]

SuperAgent/app/services/router.py

@@ -0,0 +1,881 @@
+import os
+import re
+import logging
+import httpx
+import datetime
+from typing import Any, Literal, TypedDict, Optional, List, Dict
+from langgraph.graph import END, START, StateGraph
+from ..models.schemas import ChatRequest, ChatResponse
+from groq import Groq
+from .asset_registry import asset_registry
+
+logger = logging.getLogger(__name__)
+
+Intent = Literal["live", "analysis", "anomaly", "report"]
+
+
+class SuperAgentState(TypedDict, total=False):
+    message: str          # The actual question (after stripping the User_ID prefix)
+    raw_message: str      # Original message exactly as received
+    session_id: str
+    user_id: str
+    user_authorized: bool # True once permission check passes
+    check_all: bool
+    force_retrain: bool
+    asset_id: str | None
+    asset_name: str | None
+    agent_calls: list[str]
+    intent: Intent
+    answer: str
+    analysis: Any
+    data: list[dict[str, Any]]
+    generated_sql: str | None
+    orchestration: list[str]
+    report_format: Literal["excel", "pdf"]
+    schema_name: str | None
+
+
+class RouterService:
+    def __init__(self):
+        api_key = os.getenv("GROQ_API_KEY")
+        self.client = Groq(api_key=api_key) if api_key else None
+        self.model = os.getenv("GROQ_MODEL", "llama-3.3-70b-versatile")
+        
+        # Hugging Face uses port 7860, locally we default to 8000
+        port = os.getenv("PORT", "8000")
+        
+        self.live_agent_url = os.getenv("LIVE_AGENT_URL", f"http://127.0.0.1:{port}/live")
+        self.analysis_agent_url = os.getenv("ANALYSIS_AGENT_URL", f"http://127.0.0.1:{port}/analysis")
+        self.anomaly_agent_url = os.getenv("ANOMALY_AGENT_URL", f"http://127.0.0.1:{port}/anomaly")
+        self.telemetry_agent_url = os.getenv("TELEMETRY_AGENT_URL", f"http://127.0.0.1:{port}/telemetry")
+        
+        # Bulletproof port rewrite for any loopback URLs
+        for attr in ["live_agent_url", "analysis_agent_url", "anomaly_agent_url", "telemetry_agent_url"]:
+            val = getattr(self, attr)
+            if "localhost" in val or "127.0.0.1" in val:
+                new_val = re.sub(r":\d+", f":{port}", val)
+                setattr(self, attr, new_val)
+                
+        self.graph = self._build_graph()
+
+    def _build_graph(self):
+        graph = StateGraph(SuperAgentState)
+
+        # --- Permission gate (runs first) ---
+        graph.add_node("check_permissions", self._check_permissions_node)
+        graph.add_node("classify_intent", self._classify_node)
+        graph.add_node("route_live", self._route_live_node)
+        graph.add_node("route_analysis", self._route_analysis_node)
+        graph.add_node("route_anomaly", self._route_anomaly_node)
+        graph.add_node("route_report_intent", self._route_report_intent_node)
+        graph.add_node("generate_excel", self._generate_excel_node)
+        graph.add_node("generate_pdf", self._generate_pdf_node)
+
+        graph.add_edge(START, "check_permissions")
+        graph.add_conditional_edges(
+            "check_permissions",
+            self._route_after_permission_check,
+            {
+                "authorized": "classify_intent",
+                "denied":     END,       # Short-circuit — answer already set
+            },
+        )
+        graph.add_conditional_edges(
+            "classify_intent",
+            self._route_from_intent,
+            {
+                "live": "route_live",
+                "analysis": "route_analysis",
+                "anomaly": "route_anomaly",
+                "report": "route_report_intent",
+            },
+        )
+        graph.add_conditional_edges(
+            "route_report_intent",
+            self._route_from_report_format,
+            {
+                "excel": "generate_excel",
+                "pdf": "generate_pdf",
+            },
+        )
+        graph.add_edge("route_live", END)
+        graph.add_edge("route_analysis", END)
+        graph.add_edge("route_anomaly", END)
+        graph.add_edge("generate_excel", END)
+        graph.add_edge("generate_pdf", END)
+        return graph.compile()
+
+    async def classify_intent(self, message: str) -> Intent:
+        deterministic_intent = self._classify_with_rules(message)
+        if deterministic_intent is not None:
+            return deterministic_intent
+
+        if self.client is None:
+            return "live"
+
+        prompt = f"""
+        Classify this telemetry question for routing.
+
+        Return "report" when the user wants to generate, download, or create a report file,
+        especially when they mention downloading, exporting, or creating an excel, xlsx,
+        pdf, sheet, or document, or ask to "generate a report" for a specific asset.
+
+        Return "analysis" when the user asks for historical analysis, trend analysis,
+        reports, charts, graphs, summaries over a time period, comparisons,
+        averages/min/max over a window, scheduled analysis, or any
+        date/range-based request.
+        
+        Return "anomaly" when the user asks for status, anomalies, faults, incidents, 
+        spikes, drops, mentioned "foreman", or asks about coldrooms/tanks/smoke alarm system status,
+        what is happening, or how they are doing (e.g. fire safety, abnormal behavior).
+
+        Return "live" when the user asks for current/latest/realtime status,
+        current value, online/offline state, latest sensor reading, current tank
+        level, live alerts, or what is happening right now.
+
+        User Message: "{message}"
+
+        Return ONLY the word "live", "analysis", "anomaly", or "report".
+        """
+        
+        response = self.client.chat.completions.create(
+            messages=[
+                {"role": "user", "content": prompt}
+            ],
+            model=self.model,
+            temperature=0,
+            max_tokens=10,
+        )
+        
+        intent = response.choices[0].message.content.strip().lower()
+        if "anomaly" in intent:
+            return "anomaly"
+        if "analysis" in intent:
+            return "analysis"
+        if "report" in intent:
+            return "report"
+        return "live"
+
+    # ------------------------------------------------------------------
+    # Permission-check node
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _parse_user_message(raw: str) -> tuple[str, str]:
+        """
+        Splits a message of the form  ``User_ID: question``.
+
+        Returns (user_id_or_empty, question).
+        The separator is the *first* colon that follows a non-whitespace
+        token at the start of the string so that questions containing colons
+        are never accidentally split.
+        """
+        # Match  <anything without whitespace>  :  <rest>
+        m = re.match(r'^(\S+?)\s*:\s*(.+)$', raw.strip(), re.DOTALL)
+        if m:
+            return m.group(1).strip(), m.group(2).strip()
+        return "", raw.strip()
+
+    async def _check_permissions_node(self, state: SuperAgentState) -> SuperAgentState:
+        """
+        First node in the graph.
+
+        1. Parses ``User_ID: question`` out of raw_message.
+        2. Resolves the real user UUID (falls back to DEFAULT_USER_ID when
+           the parsed prefix looks like the legacy placeholder ``user_1``).
+        3. Loads the user's allowed assets from the registry.
+        4. Blocks the request (sets ``user_authorized=False``) when the user
+           has zero assets, indicating they don't exist in the system.
+        """
+        raw = state.get("raw_message") or state.get("message", "")
+        default_uid = os.getenv("DEFAULT_USER_ID", "ad7103e4-9595-4000-a56a-8f650e3eddef")
+
+        parsed_uid, question = self._parse_user_message(raw)
+        orch = state.get("orchestration", [])
+
+        # Resolve real UUID
+        if parsed_uid and parsed_uid not in ("", "user_1"):
+            user_id = parsed_uid
+            orch.append(f"Gateway: Parsed User_ID='{user_id}' from message prefix")
+        else:
+            user_id = state.get("user_id")
+            if not parsed_uid:
+                orch.append("Gateway: No User_ID prefix found — using session/default user")
+
+        # Check permissions via asset registry
+        allowed_assets = asset_registry.list_all_assets(user_id)
+
+        if not allowed_assets:
+            logger.warning(f"Permission denied: user '{user_id}' has no authorized assets.")
+            orch.append(f"Gateway: ❌ Permission denied — user '{user_id}' not found or has no assets.")
+            return {
+                "user_id": user_id,
+                "message": question,
+                "raw_message": raw,
+                "user_authorized": False,
+                "orchestration": orch,
+                "answer": (
+                    f"⛔ Access denied. The user **`{user_id}`** does not exist in the system "
+                    "or has no assets assigned. Please verify your User ID and try again."
+                ),
+                "intent": "live",
+            }
+
+        orch.append(
+            f"Gateway: ✅ User '{user_id}' authorized — {len(allowed_assets)} asset(s) available."
+        )
+        return {
+            "user_id": user_id,
+            "message": question,
+            "raw_message": raw,
+            "user_authorized": True,
+            "orchestration": orch,
+        }
+
+    def _route_after_permission_check(self, state: SuperAgentState) -> str:
+        return "authorized" if state.get("user_authorized", False) else "denied"
+
+    # ------------------------------------------------------------------
+    # Intent classification node
+    # ------------------------------------------------------------------
+
+    async def _classify_node(self, state: SuperAgentState) -> SuperAgentState:
+        intent = await self.classify_intent(state["message"])
+        
+        is_all_query = any(kw in state["message"].lower() for kw in ["all assets", "show assets", "list assets", "summary"])
+        
+        asset_id = None
+        asset_name = None
+        
+        # 1. First Priority: Try to resolve from the message query itself
+        if not is_all_query:
+            asset = asset_registry.find_asset(state["message"], state["user_id"])
+            if asset:
+                asset_id = asset['id']
+                asset_name = asset['name']
+        
+        orch = state.get("orchestration", [])
+        orch.append(f"Gateway: Intent classified as '{intent}'")
+        
+        if is_all_query:
+            summary = asset_registry.get_summary(state["user_id"])
+            orch.append(f"Gateway: 'All Assets' query detected. System Summary: {summary}")
+        elif asset_id:
+            orch.append(f"Gateway: Identified asset '{asset_name}' ({asset_id})")
+
+        return {
+            "intent": intent or "live", 
+            "asset_id": asset_id, 
+            "asset_name": asset_name, 
+            "orchestration": orch
+        }
+
+    def _route_from_intent(self, state: SuperAgentState) -> Intent:
+        return state.get("intent", "live")
+
+    async def _route_live_node(self, state: SuperAgentState) -> SuperAgentState:
+        orch = state.get("orchestration", [])
+        orch.append("Gateway -> Live Agent: Fetching real-time telemetry from database")
+        live_payload = {
+            "session_id": state.get("session_id") or "default",
+            "user_id": state.get("user_id"),
+            "message": state["message"],
+            "asset_id": state.get("asset_id"),
+            "asset_name": state.get("asset_name"),
+            "schema_name": state.get("schema_name"),
+        }
+        async with httpx.AsyncClient(timeout=300.0) as client:
+            response = await client.post(f"{self.telemetry_agent_url}/api/v1/chat/message", json=live_payload)
+            if response.status_code != 200:
+                orch.append("Live Agent: Failed to respond")
+                print(f"\n [GATEWAY ERROR] Live Route Failed: {response.text}\n")
+                return {
+                    # "answer": f"Live Agent Error: {response.text}",
+                    "answer": "The Gateway encountered an error communicating with the live telemetry service. Please try again.",
+                    "intent": "live",
+                    "orchestration": orch
+                }
+            data = response.json()
+            orch.append("Live Agent: Response received successfully")
+
+        return {
+            "answer": data.get("answer", ""),
+            "intent": "live",
+            "data": data.get("rows", []),
+            "generated_sql": data.get("generated_sql"),
+            "orchestration": orch
+        }
+
+    async def _route_analysis_node(self, state: SuperAgentState) -> SuperAgentState:
+        orch = state.get("orchestration", [])
+        orch.append("Gateway -> Analysis Agent: Initiating historical trend analysis")
+        analysis_payload = {
+            "message": state["message"],
+            "user_id": state.get("user_id"),
+            "asset_id": state.get("asset_id"),
+            "asset_name": state.get("asset_name"),
+            "schema_name": state.get("schema_name"),
+        }
+        async with httpx.AsyncClient(timeout=300.0) as client:
+            response = await client.post(
+                f"{self.telemetry_agent_url}/api/v1/chat/message",
+                json=analysis_payload,
+            )
+            if response.status_code != 200:
+                orch.append("Analysis Agent: Failed to respond")
+                print(f"\n [GATEWAY ERROR] Analysis Route Failed: {response.text}\n")
+                return {
+                    # "answer": f"Analysis Agent Error: {response.text}",
+                    "answer": "The Gateway encountered an error communicating with the analysis service. Please try again.",
+                    "intent": "analysis",
+                    "orchestration": orch
+                }
+            data = response.json()
+            orch.append("Analysis Agent: Trend analysis complete")
+
+        return {
+            "answer": data.get("answer", ""),
+            "intent": "analysis",
+            "analysis": data.get("analysis"),
+            "orchestration": orch
+        }
+
+    async def _route_anomaly_node(self, state: SuperAgentState) -> SuperAgentState:
+        orch = state.get("orchestration", [])
+        orch.append("Gateway -> Anomaly Agent (The Foreman): Activating multi-agent industrial supervisor")
+        anomaly_payload = {
+            "message": state["message"],
+            "user_id": state.get("user_id"),
+            "asset_id": state.get("asset_id"),
+            "asset_name": state.get("asset_name"),
+            "schema_name": state.get("schema_name"),
+        }
+        async with httpx.AsyncClient(timeout=600.0) as client:
+            response = await client.post(
+                f"{self.anomaly_agent_url}/api/chat",
+                json=anomaly_payload,
+            )
+            if response.status_code != 200:
+                orch.append("Anomaly Agent: Failed to respond")
+                return {
+                    "answer": f"Anomaly Agent Error: {response.text}",
+                    "intent": "anomaly",
+                    "orchestration": orch
+                }
+            data = response.json()
+            orch.append("Anomaly Agent: Multi-agent reasoning cycle finished")
+
+        return {
+            "answer": data.get("reply", ""),
+            "intent": "anomaly",
+            "data": data.get("data"),
+            "orchestration": orch
+        }
+
+    def _classify_with_rules(self, message: str) -> Intent | None:
+        text = self._normalize_text(message)
+
+        # Check for report generation intent
+        report_keywords = [
+            "generate report", "download report", "get report", "create report",
+            "excel report", "pdf report", "xlsx report", "sheet report", "spreadsheet report",
+            "generate excel", "generate pdf", "download excel", "download pdf",
+            "save as pdf", "save as excel", "export to pdf", "export to excel",
+            "export pdf", "export excel", "make pdf", "make excel", "telemetry pdf", "telemetry excel",
+            "pdf file", "excel file", "xlsx file", "spreadsheet file", "sheet file",
+            "pdf", "excel", "xlsx", "spreadsheet", "csv"
+        ]
+        if any(kw in text for kw in report_keywords):
+            return "report"
+
+        # Route coldrooms/tanks/smoke alarm system status, what is happening, 
+        # and how they are doing to "anomaly" so that the anomaly agent is run.
+        has_specific_asset = any(term in text for term in [
+            "coldroom", "cold room", "coldrooms", "cold rooms",
+            "tank", "tanks",
+            "smoke alarm", "smoke alarms", "smoke detector", "smoke detectors",
+            "smoke sensor", "smoke sensors", "fire alarm", "fire alarms"
+        ])
+        if has_specific_asset:
+            is_status_query = "status" in text or "state" in text
+            is_happening_query = "happening" in text or "going on" in text or "what is with" in text
+            is_doing_query = (
+                "doing" in text or 
+                "how is" in text or 
+                "how are" in text or 
+                "how's" in text or
+                "is ok" in text or 
+                "is okay" in text or 
+                "are ok" in text or 
+                "are okay" in text
+            )
+            if is_status_query or is_happening_query or is_doing_query:
+                is_global = any(w in text for w in [
+                    "all", "every", "scan", "system", "saare", "incidents",
+                    "tanks", "coldrooms", "zones", "alarms", "detectors", "sensors"
+                ])
+                has_number = bool(re.search(r'\d+', text))
+                if is_global or has_number:
+                    return "anomaly"
+
+        # Force "live" routing for global asset status summaries
+        global_status_keywords = ["systemstatus", "statusofall", "allassetsstatus", "alldevicesstatus", "completestatus", "statusreport"]
+        if any(kw in text.replace(" ", "") for kw in global_status_keywords):
+            return "live"
+
+        analysis_phrases = (
+            "analysis",
+            "analyze",
+            "analyse",
+            "trend",
+            "history",
+            "historical",
+            "report",
+            "graph",
+            "chart",
+            "compare",
+            "comparison",
+            "average",
+            "avg",
+            "minimum",
+            "maximum",
+            "variance",
+            "stability",
+            "stable",
+            "performance",
+            "efficiency",
+            "downtime",
+            "summary",
+            "summarize",
+            "scheduled analysis",
+            "schedule analysis",
+            "scheduled",
+            "forecast",
+        )
+        anomaly_phrases = (
+            "anomaly",
+            "anomalies",
+            "fault",
+            "faults",
+            "incident",
+            "incidents",
+            "spike",
+            "drop",
+            "suspicious",
+            "smoke",
+            "fire",
+            "foreman",
+            "abnormal",
+            "behavior",
+            "behaviour",
+            "faults",
+            "incidents",
+            "spikes",
+            "active alerts"
+        )
+        live_phrases = (
+            "right now",
+            "at the moment",
+            "currently",
+            "current",
+            "realtime",
+            "real time",
+            "real-time",
+            "live",
+            "latest",
+            "now",
+            "online",
+            "offline",
+            "status",
+            "state",
+            "value",
+            "reading",
+            "level",
+            "alert",
+            "alerts",
+            "show all",
+            "all assets",
+            "all devices",
+            "all attributes",
+            "end attributes",
+            "category",
+            "category assets",
+            "category devices",
+        )
+        time_window_patterns = (
+            r"\blast\s+\d+\s*(minute|minutes|hour|hours|day|days|week|weeks|month|months)\b",
+            r"\bpast\s+\d+\s*(minute|minutes|hour|hours|day|days|week|weeks|month|months)\b",
+            r"\bprevious\s+\d+\s*(minute|minutes|hour|hours|day|days|week|weeks|month|months)\b",
+            r"\bfor\s+(today|yesterday|tomorrow)\b",
+            r"\b(today|yesterday)\b",
+            r"\bbetween\b.+\band\b",
+            r"\bfrom\b.+\bto\b",
+            r"\bsince\b",
+            r"\bover\s+the\s+(last|past|previous)\b",
+            r"\b\d{1,2}(st|nd|rd|th)?\s+(jan|feb|mar|apr|may|jun|jul|aug|sep|sept|oct|nov|dec)[a-z]*\s+\d{4}\b",
+            r"\b\d{4}-\d{1,2}-\d{1,2}\b",
+            r"\b\d{1,2}/\d{1,2}/\d{2,4}\b",
+        )
+
+        has_analysis_phrase = any(phrase in text for phrase in analysis_phrases)
+        has_anomaly_phrase = any(phrase in text for phrase in anomaly_phrases)
+        has_live_phrase = any(phrase in text for phrase in live_phrases)
+        has_time_window = any(re.search(pattern, text) for pattern in time_window_patterns)
+
+        if has_anomaly_phrase:
+            return "anomaly"
+        if has_analysis_phrase or has_time_window:
+            return "analysis"
+        if has_live_phrase:
+            return "live"
+        return None
+
+    async def route_request(self, payload: ChatRequest) -> ChatResponse:
+        raw_uid = payload.user_id or ""
+        session_user_id = raw_uid if (raw_uid and raw_uid != "user_1") else None
+
+        parsed_uid, question = self._parse_user_message(payload.message)
+
+        if self._is_greeting(question):
+            # Intercept and return a polite, friendly greeting directly without database query
+            greeting_answer = await self._generate_greeting_response(question)
+            orch_steps = ["Gateway: Intercepted greeting message — responded directly without database query."]
+            print("\n🔄 System Orchestration:\n" + "\n".join([f"- {step}" for step in orch_steps]))
+            return ChatResponse(
+                answer=greeting_answer,
+                intent="live",
+                orchestration=None
+            )
+
+        # Centralized schema verification (Hardcoded to openremote to clean up JSON API)
+        schema_name = "openremote"
+
+        final_state = await self.graph.ainvoke(
+            {
+                # Pass the raw message untouched so the permission node can parse the prefix
+                "raw_message": payload.message,
+                "message": payload.message,   # will be replaced by the permission node
+                "asset_id": payload.asset_id,
+                "session_id": payload.session_id or "default",
+                "user_id": session_user_id,   # may be overridden by parsed prefix
+                "schema_name": schema_name,
+                "orchestration": []
+            }
+        )
+        
+        answer = final_state.get("answer", "")
+        orch_steps = final_state.get("orchestration", [])
+        if orch_steps:
+            print("\n🔄 System Orchestration:\n" + "\n".join([f"- {step}" for step in orch_steps]))
+
+        return ChatResponse(
+            answer=answer,
+            intent=final_state.get("intent", "live"),
+            analysis=final_state.get("analysis"),
+            data=final_state.get("data"),
+            generated_sql=final_state.get("generated_sql"),
+            orchestration=None
+        )
+
+
+    def _is_greeting(self, message: str) -> bool:
+        # Normalize: convert to lower case, strip extra whitespace, and remove punctuation at start/end
+        text = message.strip().lower()
+        text_clean = re.sub(r"^[^\w]+|[^\w]+$", "", text)  # remove non-alphanumeric at start/end
+        
+        # Split into words to analyze length and content
+        words = text_clean.split()
+        if not words:
+            return False
+            
+        # Common greeting words/phrases
+        greeting_words = {
+            "hi", 'hii','hiii', "hy", "hello", "hey", "hola", "yo", "greetings", "howdy", "helo", "hii", "heyy",
+            "salam", "assalamualaikum", "namaste", "wassup", "sup",
+            "ok", "okay", "thanks", "thankyou", "thank you", "nice", "good", "great", "fine", "cool", "perfect"
+        }
+        
+        greeting_phrases = [
+            "good morning", "good afternoon", "good evening", "as salam alaykum", "whats up", "what's up",
+            "hi there", "hello there", "hey there", "how are you", "how r u", "how is it going", "how's it going",
+            "how are you today", "how are you doing", "hope you are doing well"
+        ]
+        
+        # Check if the entire message is one of the greeting phrases
+        if any(phrase in text_clean for phrase in greeting_phrases):
+            # Check for telemetry keywords to prevent false positives
+            if not any(kw in text_clean for kw in self._get_telemetry_keywords()):
+                return True
+                
+        # Check if the first word or the only word is a greeting word
+        # If it's a single-word message and it matches a greeting word, it's a greeting
+        if len(words) == 1 and words[0] in greeting_words:
+            return True
+            
+        # If it's a short message (e.g. up to 4 words) starting with a greeting word and contains no telemetry keywords, it's a greeting
+        if len(words) <= 4 and words[0] in greeting_words:
+            if not any(kw in text_clean for kw in self._get_telemetry_keywords()):
+                return True
+                
+        return False
+        
+    def _get_telemetry_keywords(self) -> set[str]:
+        return {
+            "asset", "device", "tank", "coldroom", "cold room", "temperature", "temp", "oil", "level", "value",
+            "status", "online", "offline", "fault", "anomaly", "anomalies", "alarm", "smoke", "fire", "incident", "incidents",
+            "report", "graph", "chart", "analyze", "analysis", "trend", "history", "historical", "reading", "humidity",
+            "zone", "refinery", "foreman", "baseline", "sensor", "attribute", "attributes", "show all", "list", "query"
+        }
+
+    async def _generate_greeting_response(self, greeting: str) -> str:
+        if self.client:
+            try:
+                prompt = f"""
+                You are the Multi-Agent Telemetry SuperAgent. The user has greeted you with: "{greeting}".
+                Respond to their greeting in an extremely polite, warm, and professional way.
+                Keep it brief (2-3 sentences), and offer to help them monitor their IoT assets, analyze telemetry, or check for anomalies.
+                Do not query any databases or mention any specific asset data.
+                """
+                response = self.client.chat.completions.create(
+                    messages=[{"role": "user", "content": prompt}],
+                    model=self.model,
+                    temperature=0.7,
+                    max_tokens=150,
+                )
+                return response.choices[0].message.content.strip()
+            except Exception as e:
+                logger.error(f"Error generating greeting response with LLM: {e}")
+                
+        # Fallback static response if LLM is unavailable or fails
+        return (
+            "Hello! I am your Multi-Agent Telemetry SuperAgent. 🌟\n\n"
+            "I'm here to help you monitor and analyze your IoT assets (like coldrooms and tanks) in real-time, "
+            "generate historical trend reports, or scan for anomalies with the Foreman.\n\n"
+            "How can I assist you today? Feel free to ask me questions like:\n"
+            "- *'What is the current status of all my assets?'*\n"
+            "- *'Compare the performance of tank 1 and tank 2 for yesterday.'*\n"
+            "- *'Are there any fire or smoke incidents in zone 3?'*"
+        )
+
+    async def _route_report_intent_node(self, state: SuperAgentState) -> SuperAgentState:
+        from .report_generator import report_generator_service
+        orch = state.get("orchestration", [])
+        fmt = report_generator_service.parse_format(state["message"])
+        orch.append(f"Report Agent: Identified format request as '{fmt.upper()}'")
+        return {
+            "report_format": fmt,
+            "orchestration": orch
+        }
+
+    def _route_from_report_format(self, state: SuperAgentState) -> str:
+        return state.get("report_format", "pdf")
+
+    async def _analyze_and_enrich_report(self, state: SuperAgentState) -> tuple[Optional[str], Optional[list[dict]], list[str]]:
+        """
+        Extracts the core query, runs the appropriate underlying agent (live, analysis, anomaly)
+        to get actual analytical findings and telemetry rows, and returns:
+        (agent_answer, telemetry_rows, orchestration_steps)
+        """
+        orch = state.get("orchestration", [])
+        original_msg = state["message"]
+        
+        # 1. Extract core query
+        core_query = self._extract_core_query(original_msg)
+        orch.append(f"Report Agent: Extracted core query for analysis: '{core_query}'")
+        
+        # 2. Classify intent of the core query
+        underlying_intent = await self.classify_intent(core_query)
+        if underlying_intent == "report":
+            # Fallback rules to avoid infinite loop
+            text = self._normalize_text(core_query)
+            if any(phrase in text for phrase in ["anomaly", "fault", "incident", "status", "doing"]):
+                underlying_intent = "anomaly"
+            elif any(phrase in text for phrase in ["trend", "history", "average", "avg", "last", "past"]):
+                underlying_intent = "analysis"
+            else:
+                underlying_intent = "live"
+                
+        orch.append(f"Report Agent: Core query classified as '{underlying_intent}' intent")
+        
+        agent_answer = None
+        telemetry_rows = None
+        
+        # 3. Call the appropriate underlying agent
+        if underlying_intent == "live":
+            orch.append("Report Agent -> Routing core query to Live Agent...")
+            live_payload = {
+                "session_id": state.get("session_id") or "default",
+                "user_id": state.get("user_id"),
+                "message": core_query,
+                "asset_id": state.get("asset_id"),
+                "asset_name": state.get("asset_name"),
+                "schema_name": state.get("schema_name"),
+            }
+            try:
+                async with httpx.AsyncClient(timeout=300.0) as client:
+                    response = await client.post(f"{self.telemetry_agent_url}/api/v1/chat/message", json=live_payload)
+                    if response.status_code == 200:
+                        data = response.json()
+                        agent_answer = data.get("answer", "")
+                        telemetry_rows = data.get("rows", [])
+                        orch.append("Report Agent: Successfully retrieved real-time analysis and data from Live Agent")
+                    else:
+                        orch.append(f"Report Agent: Live Agent responded with status {response.status_code}")
+            except Exception as e:
+                orch.append(f"Report Agent: Error calling Live Agent: {str(e)}")
+                
+        elif underlying_intent == "analysis":
+            orch.append("Report Agent -> Routing core query to Analysis Agent...")
+            analysis_payload = {
+                "message": core_query,
+                "user_id": state.get("user_id"),
+                "asset_id": state.get("asset_id"),
+                "asset_name": state.get("asset_name"),
+                "schema_name": state.get("schema_name"),
+            }
+            try:
+                async with httpx.AsyncClient(timeout=300.0) as client:
+                    response = await client.post(f"{self.telemetry_agent_url}/api/v1/chat/message", json=analysis_payload)
+                    if response.status_code == 200:
+                        data = response.json()
+                        agent_answer = data.get("answer", "")
+                        # Try to extract data if present in analysis
+                        analysis_data = data.get("analysis", {})
+                        if isinstance(analysis_data, dict):
+                            series = analysis_data.get("trend_chart", {}).get("series", [])
+                            if series and isinstance(series, list):
+                                rows = []
+                                for s in series:
+                                    metric = s.get("label", "Telemetry")
+                                    for p in s.get("points", []):
+                                        rows.append({
+                                            "timestamp": p.get("label", "N/A"),
+                                            "metric": metric,
+                                            "value": p.get("value", 0),
+                                            "unit": "Units",
+                                            "status": "NORMAL"
+                                        })
+                                if rows:
+                                    telemetry_rows = rows
+                        orch.append("Report Agent: Successfully retrieved trend analysis from Analysis Agent")
+                    else:
+                        orch.append(f"Report Agent: Analysis Agent responded with status {response.status_code}")
+            except Exception as e:
+                orch.append(f"Report Agent: Error calling Analysis Agent: {str(e)}")
+                
+        elif underlying_intent == "anomaly":
+            orch.append("Report Agent -> Routing core query to Anomaly Agent...")
+            anomaly_payload = {
+                "message": core_query,
+                "user_id": state.get("user_id"),
+                "asset_id": state.get("asset_id"),
+                "asset_name": state.get("asset_name"),
+                "schema_name": state.get("schema_name"),
+            }
+            try:
+                async with httpx.AsyncClient(timeout=600.0) as client:
+                    response = await client.post(f"{self.anomaly_agent_url}/api/chat", json=anomaly_payload)
+                    if response.status_code == 200:
+                        data = response.json()
+                        agent_answer = data.get("reply", "")
+                        anomaly_data = data.get("data", [])
+                        if anomaly_data and isinstance(anomaly_data, list):
+                            telemetry_rows = []
+                            for item in anomaly_data:
+                                telemetry_rows.append({
+                                    "timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+                                    "metric": "Anomaly Status",
+                                    "value": item.get("value") or item.get("latest_mse", "N/A"),
+                                    "unit": "Status",
+                                    "status": item.get("status") or "CRITICAL"
+                                })
+                        orch.append("Report Agent: Successfully retrieved anomaly diagnosis from Anomaly Agent")
+                    else:
+                        orch.append(f"Report Agent: Anomaly Agent responded with status {response.status_code}")
+            except Exception as e:
+                orch.append(f"Report Agent: Error calling Anomaly Agent: {str(e)}")
+                
+        return agent_answer, telemetry_rows, orch
+
+    def _extract_core_query(self, message: str) -> str:
+        msg = message
+        patterns = [
+            r"\b(?:generate|download|get|create|save|export|make|show)\s+(?:a\s+)?(?:stylized|premium|corporate|operational\s+)?(?:pdf|excel|xlsx|spreadsheet|csv|sheet|file|document)?(?:\s+report)?\s*(?:for|of|on|about)?\b",
+            r"\b(?:pdf|excel|xlsx|spreadsheet|csv|sheet|file|document)\s+report\s*(?:for|of|on|about)?\b",
+            r"\b(?:pdf|excel|xlsx|spreadsheet|csv|sheet|file|document)\s*(?:please|plz)?\b",
+            r"\b(?:please|plz)\s*(?:send|give|generate|download)\b",
+        ]
+        for pattern in patterns:
+            msg = re.sub(pattern, "", msg, flags=re.IGNORECASE)
+        msg = re.sub(r"\s+", " ", msg).strip()
+        msg = re.sub(r"^[?,.:;\s]+|[?,.:;\s]+$", "", msg)
+        return msg if msg else message
+
+    async def _generate_excel_node(self, state: SuperAgentState) -> SuperAgentState:
+        from .report_generator import report_generator_service
+        orch = state.get("orchestration", [])
+        
+        # Pre-analyze core query
+        agent_answer, telemetry_rows, orch = await self._analyze_and_enrich_report(state)
+        
+        orch.append("Report Agent -> Excel Generator: Compiling custom XLSX operational sheet")
+        res = report_generator_service.generate_report(
+            user_id=state.get("user_id"),
+            asset_query=state["message"],
+            format_type="excel",
+            custom_data=telemetry_rows,
+            executive_summary=agent_answer,
+            asset_id=state.get("asset_id")
+        )
+        orch.append("Excel Generator: File compiled successfully and link created")
+        
+        final_answer = ""
+        if agent_answer:
+            clean_ans = agent_answer
+            if "**System Orchestration:**" in clean_ans:
+                clean_ans = clean_ans.split("**System Orchestration:**")[0].strip()
+            final_answer += f"### 🔍 Analytical Findings & Diagnosis\n{clean_ans}\n\n---\n\n"
+        final_answer += res["answer"]
+        
+        return {
+            "answer": final_answer,
+            "intent": "report",
+            "orchestration": orch
+        }
+
+    async def _generate_pdf_node(self, state: SuperAgentState) -> SuperAgentState:
+        from .report_generator import report_generator_service
+        orch = state.get("orchestration", [])
+        
+        # Pre-analyze core query
+        agent_answer, telemetry_rows, orch = await self._analyze_and_enrich_report(state)
+        
+        orch.append("Report Agent -> PDF Generator: Generating stylized operational report document")
+        res = report_generator_service.generate_report(
+            user_id=state.get("user_id"),
+            asset_query=state["message"],
+            format_type="pdf",
+            custom_data=telemetry_rows,
+            executive_summary=agent_answer,
+            asset_id=state.get("asset_id")
+        )
+        orch.append("PDF Generator: PDF generated successfully and link created")
+        
+        final_answer = ""
+        if agent_answer:
+            clean_ans = agent_answer
+            if "**System Orchestration:**" in clean_ans:
+                clean_ans = clean_ans.split("**System Orchestration:**")[0].strip()
+            final_answer += f"### 🔍 Analytical Findings & Diagnosis\n{clean_ans}\n\n---\n\n"
+        final_answer += res["answer"]
+        
+        return {
+            "answer": final_answer,
+            "intent": "report",
+            "orchestration": orch
+        }
+
+    def _normalize_text(self, message: str) -> str:
+        return re.sub(r"\s+", " ", message.strip().lower())
+
+router_service = RouterService()

SuperAgent/main.py

@@ -0,0 +1,970 @@
+import os
+import sys
+import re
+import httpx
+import logging
+from pathlib import Path
+from typing import Optional, List, Dict, Any
+
+# Mute verbose httpx and httpcore INFO logs to keep console clean
+logging.getLogger("httpx").setLevel(logging.WARNING)
+logging.getLogger("httpcore").setLevel(logging.WARNING)
+
+# Add project root and sub-agent directories to sys.path to resolve dynamic package namespaces (Trigger Reload)
+root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if root_dir not in sys.path:
+    sys.path.insert(0, root_dir)
+
+sys.path.append(os.path.join(root_dir, "agents", "live"))
+sys.path.append(os.path.join(root_dir, "agents", "analysis", "backend"))
+sys.path.append(os.path.join(root_dir, "agents", "anomaly", "backend"))
+sys.path.append(os.path.join(root_dir, "agents", "telemetry"))
+sys.path.append(os.path.join(root_dir, "agents", "report"))
+
+from fastapi import FastAPI, HTTPException, BackgroundTasks
+from fastapi.middleware.cors import CORSMiddleware
+from dotenv import load_dotenv
+
+env_path = Path(__file__).parent.parent / ".env"
+load_dotenv(dotenv_path=env_path, override=True)
+
+from app.models.schemas import ChatRequest, ChatResponse, SessionCreateRequest, SessionTitleUpdateRequest
+from app.services.router import router_service
+
+from fastapi.responses import RedirectResponse, FileResponse
+from fastapi.staticfiles import StaticFiles
+
+app = FastAPI(title="Visibility AI Multi-Agent Gateway", version="0.1.0")
+
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+def resolve_and_verify_supabase_url() -> str:
+    import socket
+    import re
+    import urllib.request
+    import json
+    import ssl
+    
+    supabase_db_url = os.getenv("SUPABASE_DB_URL")
+    if not supabase_db_url:
+        raise ValueError("CRITICAL ERROR: SUPABASE_DB_URL environment variable is missing! Chat history requires a dedicated database configuration.")
+        
+    # Clean pgbouncer parameter
+    if "pgbouncer=" in supabase_db_url:
+        supabase_db_url = re.sub(r'[?&]pgbouncer=[^&]+', '', supabase_db_url)
+        
+    # Parse project ref and password
+    project_ref = "qwtpmkinqatjylozropu"
+    supabase_url = os.getenv("SUPABASE_URL", "")
+    match = re.search(r'https://([^.]+)\.supabase\.co', supabase_url)
+    if match:
+        project_ref = match.group(1)
+        
+    password = "zaheerafzal"
+    pass_match = re.search(r'postgresql://[^:]+:([^@]+)@', supabase_db_url)
+    if pass_match:
+        password = pass_match.group(1)
+        
+    # Try importing driver
+    connect_fn = None
+    try:
+        import psycopg
+        connect_fn = psycopg.connect
+    except ImportError:
+        try:
+            import psycopg2
+            connect_fn = psycopg2.connect
+        except ImportError:
+            pass
+
+    if not connect_fn:
+        print("SUPABASE CONNECTION: No postgres driver (psycopg/psycopg2) available for pre-verification.")
+        return supabase_db_url
+
+    # Helper function to resolve DNS over HTTPS (DoH) with SSL bypass
+    def doh_resolve(hostname: str) -> str:
+        ctx = ssl._create_unverified_context()
+        # Try Google DoH
+        try:
+            url = f"https://dns.google/resolve?name={hostname}&type=A"
+            req = urllib.request.Request(url, headers={"Accept": "application/json"})
+            with urllib.request.urlopen(req, timeout=3, context=ctx) as response:
+                data = json.loads(response.read().decode())
+                if "Answer" in data:
+                    for answer in data["Answer"]:
+                        if answer.get("type") == 1:
+                            return answer.get("data")
+        except Exception as e:
+            print(f"SUPABASE CONNECTION: Google DoH failed for {hostname}: {e}")
+            
+        # Try Cloudflare DoH
+        try:
+            url = f"https://cloudflare-dns.com/dns-query?name={hostname}&type=A"
+            req = urllib.request.Request(url, headers={"Accept": "application/dns-json"})
+            with urllib.request.urlopen(req, timeout=3, context=ctx) as response:
+                data = json.loads(response.read().decode())
+                if "Answer" in data:
+                    for answer in data["Answer"]:
+                        if answer.get("type") == 1:
+                            return answer.get("data")
+        except Exception as e:
+            print(f"SUPABASE CONNECTION: Cloudflare DoH failed for {hostname}: {e}")
+        return None
+
+    # 1. Try resolving using system DNS first
+    host_match = re.search(r'@([^:/]+)', supabase_db_url)
+    if host_match:
+        configured_host = host_match.group(1)
+        resolved_ip = None
+        try:
+            resolved_ip = socket.gethostbyname(configured_host)
+        except Exception:
+            # Fallback to DoH if system DNS fails
+            print(f"SUPABASE CONNECTION: System DNS failed to resolve '{configured_host}'. Trying DNS-over-HTTPS...")
+            resolved_ip = doh_resolve(configured_host)
+            
+        if resolved_ip:
+            test_url = supabase_db_url
+            if resolved_ip != configured_host:
+                test_url = supabase_db_url.replace(f"@{configured_host}", f"@{resolved_ip}")
+            try:
+                with connect_fn(test_url, connect_timeout=3) as conn:
+                    print(f"SUPABASE CONNECTION: Connected successfully to host/IP: {resolved_ip}")
+                    return test_url
+            except Exception as e:
+                print(f"SUPABASE CONNECTION: Connection to '{resolved_ip}' failed: {e}")
+
+    # 2. Try direct connection format (db.<project>.supabase.co:5432)
+    direct_host = f"db.{project_ref}.supabase.co"
+    resolved_direct_ip = None
+    try:
+        resolved_direct_ip = socket.gethostbyname(direct_host)
+    except Exception:
+        print(f"SUPABASE CONNECTION: System DNS failed to resolve direct host '{direct_host}'. Trying DNS-over-HTTPS...")
+        resolved_direct_ip = doh_resolve(direct_host)
+        
+    # Hardcoded IP fallback for the specific project if all lookups fail
+    if not resolved_direct_ip and project_ref == "qwtpmkinqatjylozropu":
+        print(f"SUPABASE CONNECTION: All lookups failed. Using fallback IP for {project_ref}...")
+        resolved_direct_ip = "3.111.134.120"
+        
+    if resolved_direct_ip:
+        direct_url = f"postgresql://postgres:{password}@{resolved_direct_ip}:5432/postgres"
+        try:
+            with connect_fn(direct_url, connect_timeout=3) as conn:
+                print(f"SUPABASE CONNECTION: Connected successfully to direct host IP: {resolved_direct_ip}")
+                return direct_url
+        except Exception as e:
+            print(f"SUPABASE CONNECTION: Connection to direct host IP '{resolved_direct_ip}' failed: {e}")
+
+    # If all fail, fallback to original configured URL so SQLAlchemy throws a clean error
+    print("SUPABASE CONNECTION: Auto-detection failed. Falling back to configured URL.")
+    return supabase_db_url
+
+
+supabase_db_url = resolve_and_verify_supabase_url()
+if supabase_db_url.startswith("postgresql://"):
+    supabase_db_url = supabase_db_url.replace("postgresql://", "postgresql+psycopg://", 1)
+    
+SupabaseEngine = create_engine(supabase_db_url, pool_pre_ping=True)
+SupabaseSessionLocal = sessionmaker(bind=SupabaseEngine)
+print("SUPABASE CHAT DB: Configured dedicated connection pool successfully.")
+
+# Import sub-agents applications
+from agents.live.app.main import app as live_app
+from agents.analysis.backend.app.main import app as analysis_app
+from agents.anomaly.backend.api_chat import app as anomaly_app
+from agents.telemetry.main import app as telemetry_app
+from agents.report.main import app as report_app
+
+# Mount sub-agents natively on their respective sub-paths
+app.mount("/live", live_app)
+app.mount("/analysis", analysis_app)
+app.mount("/anomaly", anomaly_app)
+app.mount("/telemetry", telemetry_app)
+app.mount("/report", report_app)
+
+# Custom route to force download report files or preview in browser
+@app.get("/reports/{filename}")
+async def download_report(filename: str):
+    
+    filepath = os.path.join("scratch/reports", filename)
+    if not os.path.exists(filepath):
+        raise HTTPException(status_code=404, detail="File not found")
+    
+    if filename.endswith(".pdf"):
+        # Serve PDF inline for browser previewing in a new tab
+        return FileResponse(
+            path=filepath,
+            media_type="application/pdf",
+            headers={"Content-Disposition": f"inline; filename={filename}"}
+        )
+    else:
+        # Serve spreadsheet as a forced direct download
+        media_type = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" if filename.endswith(".xlsx") else "application/octet-stream"
+        return FileResponse(
+            path=filepath,
+            filename=filename,
+            media_type=media_type
+        )
+
+# Serve report files statically (fallback)
+os.makedirs("scratch/reports", exist_ok=True)
+app.mount("/reports", StaticFiles(directory="scratch/reports"), name="reports")
+
+
+DEFAULT_USER_ID = os.getenv("DEFAULT_USER_ID", "ad7103e4-9595-4000-a56a-8f650e3eddef")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.get("/", tags=["Root"])
+async def read_index():
+    return {
+        "status": "online",
+        "service": "Multi-Agent Telemetry SuperAgent API",
+        "integrations": {
+            "live_agent": "Connected [OK]",
+            "analysis_agent": "Connected [OK]",
+            "anomaly_agent": "Connected [OK]",
+            "postgresql_db": "Connected [OK]"
+        },
+        "api_endpoints_guide": {
+            "/health": "GET - Check server status.",
+            "/api/v1/chat/message": "POST - Unified AI chat endpoint (Live/Analysis/Anomaly).",
+            "/api/v1/assets": "GET - List authorized assets for a user.",
+            "/api/v1/schedules": "GET - List available automated report types.",
+            "/api/v1/schedules/run/{id}": "POST - Manually trigger a specific report.",
+            "/api/v1/anomalies": "GET - Real-time summary of all active incidents.",
+            "/docs": "GET - Interactive Swagger UI documentation."
+        },
+        "message": "Backend is running independently. Connect your frontend to this URL."
+    }
+
+@app.get("/health")
+def health() -> dict[str, str]:
+    return {"status": "ok"}
+
+@app.get("/api/v1/assets")
+async def list_assets(user_id: str = None):
+    from app.services.asset_registry import asset_registry
+    assets = asset_registry.list_all_assets(user_id=user_id or DEFAULT_USER_ID)
+    return assets
+
+@app.get("/api/v1/assets/{asset_id}")
+async def get_asset(asset_id: str, user_id: str = None):
+    from app.services.asset_registry import asset_registry
+    asset = asset_registry.get_asset_by_id(asset_id, user_id=user_id or DEFAULT_USER_ID)
+    if not asset:
+        raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found or not authorized.")
+    return asset
+
+@app.get("/debug/logs")
+async def get_logs():
+    logs = {"cwd": os.getcwd()}
+    for agent in ["startup", "live_agent", "analysis_agent", "anomaly_agent"]:
+        log_path = f"/app/{agent}.log"
+        if os.path.exists(log_path):
+            with open(log_path, "r") as f:
+                logs[agent] = f.read()[-2000:] # Last 2000 chars
+        else:
+            # Fallback to relative path if /app/ doesn't exist
+            rel_path = f"../{agent}.log"
+            if os.path.exists(rel_path):
+                 with open(rel_path, "r") as f:
+                    logs[agent] = f.read()[-2000:]
+            else:
+                logs[agent] = f"Log file not found at {log_path} or {rel_path}"
+    return logs
+
+def maybe_auto_title_session(session_id: str, first_query: str):
+    import os
+    import re
+    from sqlalchemy import text
+    from app.services.asset_registry import asset_registry
+    
+    check_sql = "SELECT title FROM chat_sessions WHERE id = :id"
+    try:
+        with SupabaseSessionLocal() as session:
+            title = session.execute(text(check_sql), {"id": session_id}).scalar()
+            if not title or title != "New Chat":
+                return
+            
+            api_key = os.getenv("GROQ_API_KEY")
+            model = os.getenv("GROQ_MODEL", "llama-3.3-70b-versatile")
+            if not api_key:
+                print("AUTO-TITLE: No Groq API Key found. Skipping auto-title.")
+                return
+                
+            from groq import Groq
+            client = Groq(api_key=api_key)
+            
+            prompt = f"""
+            Create a short, professional topic title of 3-5 words summarizing the user's query.
+                Focus strictly on the core asset (e.g., Coldroom 1, Tank 3) or metric (e.g., Temperature, Humidity) mentioned.
+                Do NOT output generic diagnostic messages like "Unknown or Incomplete Query" or "Vague Query".
+                If the query is too short or hard to classify, summarize the query text directly (e.g., "Query about {first_query}").
+                Do not add extra symbols, markdown, or quotes.
+            Query: "{first_query}"
+            Title:
+            """
+            
+            response = client.chat.completions.create(
+                messages=[
+                    {"role": "user", "content": prompt}
+                ],
+                model=model,
+                temperature=0.3,
+                max_tokens=15,
+            )
+            
+            new_title = response.choices[0].message.content.strip()
+            new_title = re.sub(r'^["\']|["\']$', '', new_title).strip()
+            
+            if len(new_title) > 60:
+                new_title = new_title[:57] + "..."
+                
+            if new_title:
+                update_sql = "UPDATE chat_sessions SET title = :title WHERE id = :id"
+                session.execute(text(update_sql), {"title": new_title, "id": session_id})
+                session.commit()
+                print(f"AUTO-TITLE: Updated session {session_id} title to: '{new_title}'")
+    except Exception as e:
+        print(f"AUTO-TITLE ERROR: {e}")
+
+@app.post("/api/v1/chat/message", response_model=ChatResponse)
+async def chat_message(payload: ChatRequest, background_tasks: BackgroundTasks) -> ChatResponse:
+    print(f"DEBUG: Received request: {payload}")
+    try:
+        response = await router_service.route_request(payload)
+        
+        session_id = payload.session_id
+        if session_id and session_id != "default":
+            parsed_uid, clean_message = router_service._parse_user_message(payload.message)
+            effective_user_id = payload.user_id if (payload.user_id and payload.user_id != "user_1") else DEFAULT_USER_ID
+            if parsed_uid and parsed_uid != "user_1":
+                effective_user_id = parsed_uid
+                
+            ensure_session_sql = """
+            INSERT INTO chat_sessions (id, user_id, title, created_at, updated_at)
+            VALUES (:id, :user_id, :title, NOW(), NOW())
+            ON CONFLICT (id) DO UPDATE SET updated_at = NOW();
+            """
+            
+            insert_message_sql = """
+            INSERT INTO chat_messages (session_id, user_id, role, content, created_at)
+            VALUES (:session_id, :user_id, :role, :content, NOW())
+            """
+            
+            try:
+                with SupabaseSessionLocal() as db_session:
+                    db_session.execute(text(ensure_session_sql), {
+                        "id": session_id,
+                        "user_id": effective_user_id,
+                        "title": "New Chat"
+                    })
+                    
+                    db_session.execute(text(insert_message_sql), {
+                        "session_id": session_id,
+                        "user_id": effective_user_id,
+                        "role": "user",
+                        "content": clean_message
+                    })
+                    
+                    db_session.execute(text(insert_message_sql), {
+                        "session_id": session_id,
+                        "user_id": effective_user_id,
+                        "role": "assistant",
+                        "content": response.answer
+                    })
+                    db_session.commit()
+                    
+                    background_tasks.add_task(maybe_auto_title_session, session_id, clean_message)
+            except Exception as e:
+                print(f"CHAT HISTORY LOGGING ERROR: {e}")
+                
+        return response
+    except Exception as exc:
+        import traceback
+        error_msg = f"{type(exc).__name__}: {str(exc)}"
+        tb = traceback.format_exc()
+        print(f"ERROR: {error_msg}\n{tb}")
+        raise HTTPException(status_code=500, detail=f"Internal error: {error_msg}\n{tb[:200]}") from exc
+
+@app.get("/api/v1/chat/sessions")
+async def list_chat_sessions(user_id: Optional[str] = None):
+    effective_user_id = user_id if (user_id and user_id != "user_1") else DEFAULT_USER_ID
+    query = """
+    SELECT id, title, created_at, updated_at
+    FROM chat_sessions
+    WHERE user_id = :user_id
+    ORDER BY updated_at DESC
+    """
+    from sqlalchemy import text
+    try:
+        with SupabaseSessionLocal() as session:
+            rows = session.execute(text(query), {"user_id": effective_user_id}).mappings().all()
+            return [dict(row) for row in rows]
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to fetch chat sessions: {str(e)}")
+
+@app.post("/api/v1/chat/sessions")
+async def create_chat_session(payload: SessionCreateRequest):
+    import uuid
+    from sqlalchemy import text
+    from app.services.asset_registry import asset_registry
+    
+    user_id = payload.user_id if (payload.user_id and payload.user_id != "user_1") else DEFAULT_USER_ID
+    session_id = f"session_{uuid.uuid4().hex[:12]}"
+    title = "New Chat"
+    
+    query = """
+    INSERT INTO chat_sessions (id, user_id, title, created_at, updated_at)
+    VALUES (:id, :user_id, :title, NOW(), NOW())
+    RETURNING id, title, created_at, updated_at
+    """
+    try:
+        with SupabaseSessionLocal() as session:
+            row = session.execute(text(query), {
+                "id": session_id,
+                "user_id": user_id,
+                "title": title
+            }).mappings().first()
+            session.commit()
+            return dict(row)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to create chat session: {str(e)}")
+
+@app.get("/api/v1/chat/sessions/{session_id}/messages")
+async def get_session_messages(session_id: str):
+    from sqlalchemy import text
+    from app.services.asset_registry import asset_registry
+    
+    query = """
+    SELECT role, content
+    FROM chat_messages
+    WHERE session_id = :session_id
+    ORDER BY created_at ASC, id ASC
+    """
+    try:
+        with SupabaseSessionLocal() as session:
+            rows = session.execute(text(query), {"session_id": session_id}).mappings().all()
+            return [dict(row) for row in rows]
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to fetch messages for session: {str(e)}")
+
+@app.put("/api/v1/chat/sessions/{session_id}")
+async def update_session_title(session_id: str, payload: SessionTitleUpdateRequest):
+    from sqlalchemy import text
+    from app.services.asset_registry import asset_registry
+    
+    query = """
+    UPDATE chat_sessions
+    SET title = :title, updated_at = NOW()
+    WHERE id = :session_id
+    """
+    try:
+        with SupabaseSessionLocal() as session:
+            res = session.execute(text(query), {"title": payload.title, "session_id": session_id})
+            session.commit()
+            if res.rowcount == 0:
+                raise HTTPException(status_code=404, detail="Session not found")
+            return {"success": True}
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to update session title: {str(e)}")
+
+@app.delete("/api/v1/chat/sessions/{session_id}")
+async def delete_chat_session(session_id: str):
+    from sqlalchemy import text
+    from app.services.asset_registry import asset_registry
+    
+    delete_messages_sql = "DELETE FROM chat_messages WHERE session_id = :session_id"
+    delete_session_sql = "DELETE FROM chat_sessions WHERE id = :session_id"
+    try:
+        with SupabaseSessionLocal() as session:
+            check_res = session.execute(text("SELECT 1 FROM chat_sessions WHERE id = :session_id"), {"session_id": session_id}).scalar()
+            if not check_res:
+                raise HTTPException(status_code=404, detail="Session not found")
+                
+            session.execute(text(delete_messages_sql), {"session_id": session_id})
+            session.execute(text(delete_session_sql), {"session_id": session_id})
+            session.commit()
+            return {"success": True}
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to delete chat session: {str(e)}")
+
+
+
+@app.get("/api/v1/anomalies")
+async def get_active_anomalies(user_id: str = None, schema_name: str = None):
+    # Centralized schema verification for anomalies endpoint
+    resolved_schema = "openremote"
+    if schema_name and schema_name.lower() != "openremote":
+        try:
+            from app.services.asset_registry import asset_registry
+            from sqlalchemy import text
+            with asset_registry.SessionLocal() as session:
+                res = session.execute(text(
+                    "SELECT 1 FROM information_schema.tables WHERE table_schema = :schema AND table_name = 'asset' LIMIT 1"
+                ), {"schema": schema_name})
+                if res.scalar() is not None:
+                    resolved_schema = schema_name
+        except Exception:
+            resolved_schema = "openremote"
+
+    port = os.getenv("PORT", "8000")
+    anomaly_agent_url = os.getenv("ANOMALY_AGENT_URL", f"http://127.0.0.1:{port}/anomaly")
+    if "localhost" in anomaly_agent_url or "127.0.0.1" in anomaly_agent_url:
+        anomaly_agent_url = re.sub(r":\d+", f":{port}", anomaly_agent_url)
+    async with httpx.AsyncClient(timeout=300.0) as client:
+        response = await client.get(
+            f"{anomaly_agent_url}/api/v1/anomalies",
+            params={
+                "user_id": user_id or DEFAULT_USER_ID,
+                "schema_name": resolved_schema
+            }
+        )
+        return response.json()
+
+@app.get("/api/v1/reports/daily")
+async def get_daily_reports():
+    import csv
+    import ast
+    import sys
+    csv.field_size_limit(2147483647)
+    from pathlib import Path
+    
+    reports = {"coldrooms": [], "tanks": [], "smoke_alarms": []}
+    base_dir = Path(__file__).parent.parent
+    cr_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "coldroom_reports.csv"
+    tank_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "tank_reports.csv"
+    smoke_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "smoke_reports.csv"
+    
+    def parse_csv(file_path, target_type):
+        if not file_path.exists(): return []
+        with open(file_path, mode='r', encoding='utf-8') as f:
+            reader = csv.DictReader(f)
+            all_rows = list(reader)
+        if not all_rows: return []
+        filtered_rows = [row for row in all_rows if row.get('report_type') == target_type]
+        if not filtered_rows: return []
+        latest_time = max(row['report_timestamp'] for row in filtered_rows)
+        latest_rows = [row for row in filtered_rows if row['report_timestamp'] == latest_time]
+        for row in latest_rows:
+            if 'anomalies' in row:
+                try: row['anomalies'] = ast.literal_eval(row['anomalies'])
+                except: pass
+        return latest_rows
+
+    reports["coldrooms"] = parse_csv(cr_path, "daily")
+    reports["tanks"] = parse_csv(tank_path, "daily")
+    reports["smoke_alarms"] = parse_csv(smoke_path, "daily")
+    return reports
+
+@app.get("/api/v1/reports/latest")
+async def get_latest_rolling_reports():
+    import csv
+    import ast
+    import sys
+    csv.field_size_limit(2147483647)
+    from pathlib import Path
+    
+    reports = {"coldrooms": [], "tanks": [], "smoke_alarms": []}
+    base_dir = Path(__file__).parent.parent
+    cr_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "coldroom_reports.csv"
+    tank_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "tank_reports.csv"
+    smoke_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "smoke_reports.csv"
+    
+    def parse_csv(file_path, target_type):
+        if not file_path.exists(): return []
+        with open(file_path, mode='r', encoding='utf-8') as f:
+            reader = csv.DictReader(f)
+            all_rows = list(reader)
+        if not all_rows: return []
+        filtered_rows = [row for row in all_rows if row.get('report_type') == target_type]
+        if not filtered_rows: return []
+        latest_time = max(row['report_timestamp'] for row in filtered_rows)
+        latest_rows = [row for row in filtered_rows if row['report_timestamp'] == latest_time]
+        for row in latest_rows:
+            if 'anomalies' in row:
+                try: row['anomalies'] = ast.literal_eval(row['anomalies'])
+                except: pass
+        return latest_rows
+
+    reports["coldrooms"] = parse_csv(cr_path, "rolling")
+    reports["tanks"] = parse_csv(tank_path, "rolling")
+    reports["smoke_alarms"] = parse_csv(smoke_path, "rolling")
+    return reports
+
+import uuid
+import json
+import datetime
+import time
+import threading
+from pydantic import BaseModel
+from typing import Optional
+
+class ManualReportRequest(BaseModel):
+    user_id: Optional[str] = None
+    asset_id: str
+    timeframe: str
+    email: Optional[str] = None
+    format: Optional[str] = "pdf"
+    report_type: Optional[str] = "analysis"
+
+class ScheduleSaveRequest(BaseModel):
+    user_id: Optional[str] = None
+    asset_id: str
+    frequency: str
+    schedule_time: str
+    weekly_day: Optional[str] = None
+    recipient_email: str
+    report_start_hour: int
+    report_end_hour: int
+    report_format: str = "pdf" # 'pdf' or 'excel'
+    report_type: Optional[str] = "analysis" # 'analysis', 'trend', or 'anomaly'
+    is_active: bool = True
+
+def get_supabase_client():
+    supabase_url = os.getenv("SUPABASE_URL")
+    supabase_key = os.getenv("SUPABASE_SERVICE_KEY")
+    if not supabase_url or not supabase_key:
+        raise ValueError("SUPABASE_URL or SUPABASE_SERVICE_KEY environment variables are not set")
+    from supabase import create_client
+    return create_client(supabase_url, supabase_key)
+
+def trigger_automated_report(user_id: str, asset_id: str, recipient_email: str, start_hour: int, end_hour: int, report_format: str, report_type: str = "analysis"):
+    try:
+        from app.services.asset_registry import asset_registry
+        from agents.report.app.services.report_generator import report_generator_service
+        from app.services.email_service import email_service
+        
+        # 1. Resolve asset
+        asset = asset_registry.get_asset_by_id(asset_id, user_id=user_id)
+        asset_name = asset["name"] if asset else asset_id
+        
+        # Build specific custom query suffix representing hours range
+        today_date = datetime.datetime.now().strftime("%Y-%m-%d")
+        custom_query = f"{asset_name} from {today_date} {start_hour:02d}:00:00 to {today_date} {end_hour:02d}:00:00"
+        
+        print(f"SCHEDULER: Generating custom timeframe report ({report_format}) for {asset_name}: '{custom_query}'...")
+        res = report_generator_service.generate_report(
+            user_id=user_id,
+            asset_query=custom_query,
+            format_type=report_format,
+            asset_id=asset_id,
+            report_type=report_type
+        )
+        
+        if res.get("success") and recipient_email:
+            filepath = res.get("filepath")
+            subject = f"Scheduled Operational Report: {asset_name}"
+            body_html = f"""
+            <html>
+                <body>
+                    <h2 style="color: #1A365D;">Scheduled Operational Report</h2>
+                    <p>Attached is the custom timeframe report (<b>{report_format.upper()}</b>) for <b>{asset_name}</b>.</p>
+                    <p><b>Data Timeframe:</b> {start_hour:02d}:00 to {end_hour:02d}:00 (Local Time)</p>
+                </body>
+            </html>
+            """
+            email_service.send_report(
+                recipient=recipient_email,
+                subject=subject,
+                body_html=body_html,
+                attachment_path=filepath
+            )
+            print(f"SCHEDULER: Automated report sent to {recipient_email}")
+        else:
+            print(f"SCHEDULER: Report generation failed or no email configured. Success: {res.get('success')}")
+    except Exception as e:
+        print(f"SCHEDULER ERROR: Failed to run automated report: {e}")
+
+def run_background_scheduler():
+    print("Background report scheduler loop running (Supabase integration)...")
+    
+    # 1. Run an immediate check on server startup to verify database queries and process pending reports
+    try:
+        print("SCHEDULER: Running initial database schedule check on server startup...")
+        supabase = get_supabase_client()
+        response = supabase.table("report_schedule").select("*").eq("is_active", True).execute()
+        schedules = response.data or []
+        
+        now = datetime.datetime.now()
+        today_str = now.strftime("%Y-%m-%d")
+        current_time_str = now.strftime("%H:%M")
+        current_day_name = now.strftime("%A")
+        
+        print(f"SCHEDULER: Found {len(schedules)} active schedules. Current local server time: {current_time_str} ({current_day_name})")
+        
+        for s in schedules:
+            frequency = s.get("frequency", "daily").lower()
+            schedule_time = s.get("schedule_time", "09:00")
+            weekly_day = s.get("weekly_day")
+            last_run = s.get("last_run")
+            
+            should_run = False
+            # Check if current time is past the schedule run time
+            if current_time_str >= schedule_time:
+                if frequency == "daily":
+                    if last_run != today_str:
+                        should_run = True
+                elif frequency == "weekly":
+                    if weekly_day and weekly_day.lower() == current_day_name.lower():
+                        if last_run != today_str:
+                            should_run = True
+            
+            print(f"SCHEDULER STARTUP CHECK -> Asset ID: {s.get('asset_id')}, Time: {schedule_time}, Frequency: {frequency}, Last Run: {last_run}, Match: {should_run}")
+            
+            if should_run:
+                asset_id = s.get("asset_id")
+                user_id = s.get("user_id") or "ad7103e4-9595-4000-a56a-8f650e3eddef"
+                recipient_email = s.get("recipient_email")
+                start_hour = s.get("report_start_hour", 9)
+                end_hour = s.get("report_end_hour", 15)
+                report_format = s.get("report_format", "pdf")
+                report_type = s.get("report_type", "analysis")
+                
+                print(f"SCHEDULER: Launching automated report thread for {recipient_email}...")
+                threading.Thread(
+                    target=trigger_automated_report,
+                    args=(user_id, asset_id, recipient_email, start_hour, end_hour, report_format, report_type)
+                ).start()
+                
+                # Log execution to Supabase to prevent duplicate execution today
+                supabase.table("report_schedule").update({"last_run": today_str}).eq("id", s["id"]).execute()
+    except Exception as startup_err:
+        print(f"SCHEDULER STARTUP ERROR: Failed to run initial startup check: {startup_err}")
+
+    # 2. Main infinite polling loop (sleep first, then check)
+    while True:
+        try:
+            # Sleep for 30 minutes (1800 seconds) between database checks
+            time.sleep(1800)
+            
+            print("SCHEDULER: Performing periodic database schedule check...")
+            supabase = get_supabase_client()
+            
+            # Fetch active schedules from Supabase report_schedule table
+            response = supabase.table("report_schedule").select("*").eq("is_active", True).execute()
+            schedules = response.data or []
+            
+            now = datetime.datetime.now()
+            today_str = now.strftime("%Y-%m-%d")
+            current_time_str = now.strftime("%H:%M")
+            current_day_name = now.strftime("%A")
+            
+            print(f"SCHEDULER: Periodic check running. Current local server time: {current_time_str}")
+            
+            for s in schedules:
+                frequency = s.get("frequency", "daily").lower()
+                schedule_time = s.get("schedule_time", "09:00")
+                weekly_day = s.get("weekly_day")
+                last_run = s.get("last_run")
+                
+                should_run = False
+                # If current time is equal to or past the scheduled time (HH:MM)
+                if current_time_str >= schedule_time:
+                    if frequency == "daily":
+                        if last_run != today_str:
+                            should_run = True
+                    elif frequency == "weekly":
+                        if weekly_day and weekly_day.lower() == current_day_name.lower():
+                            if last_run != today_str:
+                                should_run = True
+                
+                print(f"SCHEDULER PERIODIC CHECK -> Asset ID: {s.get('asset_id')}, Time: {schedule_time}, Match: {should_run}")
+                
+                if should_run:
+                    asset_id = s.get("asset_id")
+                    user_id = s.get("user_id") or "ad7103e4-9595-4000-a56a-8f650e3eddef"
+                    recipient_email = s.get("recipient_email")
+                    start_hour = s.get("report_start_hour", 9)
+                    end_hour = s.get("report_end_hour", 15)
+                    report_format = s.get("report_format", "pdf")
+                    report_type = s.get("report_type", "analysis")
+                    
+                    print(f"SCHEDULER: Launching automated report thread for {recipient_email}...")
+                    threading.Thread(
+                        target=trigger_automated_report,
+                        args=(user_id, asset_id, recipient_email, start_hour, end_hour, report_format, report_type)
+                    ).start()
+                    
+                    # Log execution date to Supabase to prevent duplicate execution today
+                    supabase.table("report_schedule").update({"last_run": today_str}).eq("id", s["id"]).execute()
+            
+        except Exception as e:
+            print(f"Error in background scheduler loop: {e}")
+
+def init_chat_history_tables():
+    from sqlalchemy import text
+    from app.services.asset_registry import asset_registry
+    
+    # 1. Create chat_sessions metadata table
+    create_sessions_sql = """
+    CREATE TABLE IF NOT EXISTS chat_sessions (
+        id TEXT PRIMARY KEY,
+        user_id TEXT NOT NULL,
+        title TEXT NOT NULL,
+        created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+        updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+    );
+    """
+    
+    # 2. Create index on user_id
+    create_index_sql = """
+    CREATE INDEX IF NOT EXISTS idx_chat_sessions_user ON chat_sessions(user_id);
+    """
+    
+    # 3. Create chat_messages table if not exists (fallback)
+    create_messages_sql = """
+    CREATE TABLE IF NOT EXISTS chat_messages (
+        id BIGSERIAL PRIMARY KEY,
+        session_id TEXT NOT NULL,
+        role TEXT NOT NULL,
+        content TEXT NOT NULL,
+        created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+    );
+    """
+    
+    # 4. Alter chat_messages to add user_id if not present
+    alter_messages_sql = """
+    ALTER TABLE chat_messages ADD COLUMN IF NOT EXISTS user_id TEXT;
+    """
+    
+    try:
+        with SupabaseEngine.begin() as conn:
+            conn.execute(text(create_sessions_sql))
+            conn.execute(text(create_index_sql))
+            conn.execute(text(create_messages_sql))
+            conn.execute(text(alter_messages_sql))
+        print("CHAT HISTORY SCHEMA: Initialized successfully.")
+    except Exception as e:
+        print(f"CHAT HISTORY SCHEMA ERROR: Failed to initialize: {e}")
+
+@app.on_event("startup")
+async def start_scheduler_on_startup():
+    # Initialize chat history database tables
+    init_chat_history_tables()
+    
+    try:
+        # Start scheduler loop if Supabase keys exist
+        get_supabase_client()
+        threading.Thread(target=run_background_scheduler, daemon=True).start()
+    except Exception as e:
+        print(f"SCHEDULER NOT STARTED: {e}")
+
+@app.post("/api/v1/analysis/schedule/manual")
+async def manual_report_endpoint(payload: ManualReportRequest):
+    try:
+        from app.services.asset_registry import asset_registry
+        from agents.report.app.services.report_generator import report_generator_service
+        from app.services.email_service import email_service
+        
+        user_id = payload.user_id or DEFAULT_USER_ID
+        asset = asset_registry.get_asset_by_id(payload.asset_id, user_id=user_id)
+        if not asset:
+            raise HTTPException(status_code=404, detail="Asset not found")
+            
+        asset_name = asset["name"]
+        asset_query = f"{asset_name} {payload.timeframe}"
+        
+        res = report_generator_service.generate_report(
+            user_id=user_id,
+            asset_query=asset_query,
+            format_type=payload.format or "pdf",
+            asset_id=payload.asset_id,
+            report_type=payload.report_type or "analysis"
+        )
+        
+        if not res.get("success"):
+            raise HTTPException(status_code=400, detail=res.get("answer", "Failed to generate report"))
+            
+        if payload.email:
+            filepath = res.get("filepath")
+            subject = f"Manual Telemetry Report: {asset_name}"
+            body_html = f"""
+            <html>
+                <body>
+                    <h2 style="color: #1A365D;">Manual Telemetry Report</h2>
+                    <p>Hello,</p>
+                    <p>You requested a manual report for <b>{asset_name}</b>.</p>
+                    <p><b>Timeframe:</b> {payload.timeframe.capitalize()}</p>
+                    <hr/>
+                    <p style="font-size: 11px; color: #718096;">Generated by Visbot Multi-Agent Gateway.</p>
+                </body>
+            </html>
+            """
+            email_service.send_report(
+                recipient=payload.email,
+                subject=subject,
+                body_html=body_html,
+                attachment_path=filepath
+            )
+            
+        return {
+            "success": True,
+            "download_url": res.get("download_url"),
+            "filename": res.get("filename")
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/api/v1/analysis/schedule/save")
+async def save_schedule_endpoint(payload: ScheduleSaveRequest):
+    try:
+        supabase = get_supabase_client()
+        user_id = payload.user_id or DEFAULT_USER_ID
+        
+        # Check if record exists for this asset_id
+        check_resp = supabase.table("report_schedule").select("id").eq("asset_id", payload.asset_id).execute()
+        
+        record_data = {
+            "user_id": user_id,
+            "asset_id": payload.asset_id,
+            "frequency": payload.frequency,
+            "schedule_time": payload.schedule_time,
+            "weekly_day": payload.weekly_day,
+            "recipient_email": payload.recipient_email,
+            "report_start_hour": payload.report_start_hour,
+            "report_end_hour": payload.report_end_hour,
+            "report_format": payload.report_format,
+            "report_type": payload.report_type,
+            "is_active": payload.is_active
+        }
+        
+        if check_resp.data:
+            existing_id = check_resp.data[0]["id"]
+            supabase.table("report_schedule").update(record_data).eq("id", existing_id).execute()
+        else:
+            supabase.table("report_schedule").insert(record_data).execute()
+            
+        return {"success": True, "message": "Schedule saved to Supabase successfully."}
+    except Exception as e:
+        import traceback
+        print(f"SAVE SCHEDULE ERROR: {e}\n{traceback.format_exc()}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.get("/api/v1/analysis/schedule/{asset_id}")
+async def get_schedule_endpoint(asset_id: str):
+    try:
+        supabase = get_supabase_client()
+        response = supabase.table("report_schedule").select("*").eq("asset_id", asset_id).execute()
+        if response.data:
+            return response.data[0]
+        return {"is_active": False}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

SuperAgent/scratch/debug_sql.txt

@@ -0,0 +1,18 @@
+WITH latest_attributes AS (
+  SELECT DISTINCT ON (attribute_name)
+    attribute_name,
+    value,
+    timestamp
+  FROM openremote.asset_datapoint
+  WHERE entity_id = (
+    SELECT id FROM openremote.asset
+    WHERE REPLACE(LOWER(name), ' ', '') = REPLACE(LOWER('Refinery'), ' ', '')
+    ORDER BY version DESC LIMIT 1
+  )
+  ORDER BY attribute_name, timestamp DESC
+)
+SELECT
+  (SELECT name FROM openremote.asset WHERE REPLACE(LOWER(name), ' ', '') = REPLACE(LOWER('Refinery'), ' ', '') ORDER BY version DESC LIMIT 1) AS asset_name,
+  MAX(timestamp) + INTERVAL '3 hours' AS local_time,
+  jsonb_object_agg(attribute_name, value) AS sensor_data
+FROM latest_attributes

SuperAgent/scratch/raw_llm_output.txt

@@ -0,0 +1,31 @@
+Based on the system message and available context, we can determine that the user is asking for the current readings of the assets.
+
+Since the user is asking for current readings, we can assume that the `asset_name` is not provided in the context. However, we can use the `allowed_assets` list to filter the assets.
+
+We will use Scenario O: Custom Report / Multi-Asset to generate the SQL query.
+
+Here is the plan:
+
+1. Filter the assets using the `allowed_assets` list.
+2. Use the filtered assets to generate the SQL query.
+3. Use the `ROW_NUMBER()` function to get the latest reading for each asset.
+4. Format the result as a Markdown table.
+
+Here is the SQL query:
+```sql
+WITH Latest AS (
+  SELECT a.name AS asset_name, ad.timestamp + INTERVAL '3 hours' AS local_time, 
+         (ad.value::jsonb - 'Timestamp' - 'timestamp') AS sensor_data
+  FROM openremote.asset a
+  JOIN openremote.asset_datapoint ad ON a.id = ad.entity_id
+  WHERE a.name IN (SELECT value FROM unnest(string_to_array(allowed_assets, ',')) AS value)
+  AND ad.attribute_name ILIKE 'data'
+  ORDER BY ad.timestamp DESC
+  LIMIT 1
+)
+SELECT asset_name, TO_CHAR(local_time, 'HH12:MI AM') AS time_label, sensor_data
+FROM Latest;
+```
+This query will return the current readings of the assets in the `allowed_assets` list.
+
+Note: Since the `asset_name` is not provided in the context, we are using the `allowed_assets` list to filter the assets. If the `asset_name` is provided in the context, we can use it to filter the assets instead.

SuperAgent/scratch/read_pdf.py

@@ -0,0 +1,44 @@
+import os
+import sys
+
+pdf_path = r"d:\Visibility Bots Multi Agent\super-agent\SuperAgent\scratch\reports\report_coldroom1_20260520_172140.pdf"
+
+if not os.path.exists(pdf_path):
+    print(f"Error: File not found at {pdf_path}")
+    sys.exit(1)
+
+print(f"File size: {os.path.getsize(pdf_path)} bytes")
+
+# Try to import typical PDF extraction libraries
+try:
+    import pypdf
+    print("Using pypdf:")
+    reader = pypdf.PdfReader(pdf_path)
+    for i, page in enumerate(reader.pages):
+        print(f"--- Page {i+1} ---")
+        print(page.extract_text())
+except ImportError:
+    try:
+        import PyPDF2
+        print("Using PyPDF2:")
+        reader = PyPDF2.PdfReader(pdf_path)
+        for i, page in enumerate(reader.pages):
+            print(f"--- Page {i+1} ---")
+            print(page.extract_text())
+    except ImportError:
+        try:
+            import pdfplumber
+            print("Using pdfplumber:")
+            with pdfplumber.open(pdf_path) as pdf:
+                for i, page in enumerate(pdf.pages):
+                    print(f"--- Page {i+1} ---")
+                    print(page.extract_text())
+        except ImportError:
+            print("No advanced PDF text extractor (pypdf, PyPDF2, pdfplumber) is installed.")
+            print("Reading raw PDF headers/metadata via basic binary inspection:")
+            with open(pdf_path, 'rb') as f:
+                head = f.read(1000)
+                print("Header:", head[:200])
+                f.seek(-1000, 2)
+                tail = f.read(1000)
+                print("Tail contains PDF EOF:", b"%%EOF" in tail)