"""
BizNode Telegram Bot — Smart Business Operator
===============================================

Two modes:
  OWNER  — direct access to RAG + admin commands (no onboarding, full power)
  CUSTOMER — onboarding state machine → verified → menu + RAG

Onboarding state flow (customers only):
  new → awaiting_name → awaiting_business → awaiting_email → awaiting_otp → verified

State is persisted in the dashboard database (bot_users table) via REST API.

Environment variables:
  BOT_TOKEN           - Telegram bot token (required)
  TELEGRAM_BOT_TOKEN  - fallback alias
  OWNER_TELEGRAM_ID   - Owner's numeric Telegram ID (bypass onboarding + admin cmds)
  DASHBOARD_URL       - Dashboard base URL (default: http://localhost:7777)
  OLLAMA_URL          - Ollama generate endpoint
  QDRANT_URL          - Qdrant base URL
  OLLAMA_MODEL        - LLM model name
  EMBED_MODEL         - Embedding model name
  BOT_LOG_DIR         - Log file directory (default: logs/)
  BOT_SECRET          - Shared secret header for dashboard API auth
"""

from __future__ import annotations

import asyncio
import logging
import os
import re
import subprocess
import sys
import time
from collections import defaultdict
from pathlib import Path

# Ensure project root is on sys.path so "from services.*" imports work
# even when CWD is docker/bot/ (bot_runner.py sets cwd there).
_PROJECT_ROOT = str(Path(__file__).resolve().parent.parent.parent)
if _PROJECT_ROOT not in sys.path:
    sys.path.insert(0, _PROJECT_ROOT)

import requests
from telegram import Update, InlineKeyboardMarkup, InlineKeyboardButton
from telegram.ext import (
    ApplicationBuilder,
    CallbackQueryHandler,
    CommandHandler,
    ContextTypes,
    MessageHandler,
    filters,
)

# ---------------------------------------------------------------------------
# Tier enforcement
# ---------------------------------------------------------------------------
from services.tier_manager import is_feature_allowed, get_current_tier, TierLevel

_CURRENT_TIER = get_current_tier()


async def _pro_upgrade_msg(update, context):
    """Reply with upgrade message when Basic/Trial user tries Pro feature."""
    await update.message.reply_text(
        f"⭐ This feature requires BizNode Pro.\n\n"
        f"Your current tier: {_CURRENT_TIER.name}\n"
        f"Upgrade at https://biznode.1bz.biz or use /buy_license",
    )


# Pro command names → feature registry key (for fallback handler)
_PRO_COMMANDS = {
    "wallet": "crypto_service",
    "verify_wallet": "crypto_service",
    "pay": "crypto_service",
    "transact": "crypto_service",
    "affiliate": "affiliate_service",
    "bo": "bizchannel",
    "bizbrain": "bizbrain_service",
}

# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
_LOG_DIR = os.getenv("BOT_LOG_DIR", "logs")
Path(_LOG_DIR).mkdir(parents=True, exist_ok=True)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler(os.path.join(_LOG_DIR, "bot.log"), encoding="utf-8"),
    ],
)
logger = logging.getLogger("biznode.bot")

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
BOT_TOKEN = os.getenv("BOT_TOKEN") or os.getenv("TELEGRAM_BOT_TOKEN", "")
if not BOT_TOKEN or BOT_TOKEN in ("YOUR_TOKEN", ""):
    logger.critical("BOT_TOKEN is not set. Set it in .env and restart.")
    sys.exit(1)

DASHBOARD_URL    = os.getenv("DASHBOARD_URL",    "http://localhost:7777")
OLLAMA_URL       = os.getenv("OLLAMA_URL",       "http://localhost:11434/api/generate")
OLLAMA_EMBED_URL = os.getenv("OLLAMA_EMBED_URL", "http://localhost:11434/api/embeddings")
QDRANT_URL       = os.getenv("QDRANT_URL",       "http://localhost:6333")
OLLAMA_MODEL     = os.getenv("OLLAMA_MODEL",     "qwen3.5:4b")
EMBED_MODEL      = os.getenv("EMBEDDING_MODEL",  "nomic-embed-text")
BOT_SECRET       = os.getenv("BOT_SECRET",       "")
DASH_HEADERS     = {"X-Bot-Secret": BOT_SECRET} if BOT_SECRET else {}

# LLM Gateway (nodemonitor) — primary LLM source, local Ollama as fallback
LLM_GATEWAY_URL = os.getenv("LLM_GATEWAY_URL", "http://1bz.biz:7333/llm/generate")
LLM_GATEWAY_KEY = os.getenv("LLM_GATEWAY_KEY", "Nk2xGrJ4ZyERwBa3mCH33iF_CwgWh6uRMFjKvcc8e48")
LLM_GATEWAY_MODEL = os.getenv("LLM_GATEWAY_MODEL", "qwen3:14b")
USE_LLM_GATEWAY = os.getenv("USE_LLM_GATEWAY", "true").lower() == "true"


def _call_llm_gateway(prompt: str, system: str = "", model: str | None = None) -> str | None:
    """Call the nodemonitor LLM gateway. Returns response text or None on failure."""
    if not USE_LLM_GATEWAY:
        return None
    try:
        resp = requests.post(
            LLM_GATEWAY_URL,
            json={
                "prompt": prompt,
                "system": system,
                "model": model or LLM_GATEWAY_MODEL,
                "caller_node": "biznode-bot",
            },
            headers={"x-api-key": LLM_GATEWAY_KEY},
            timeout=120,
        )
        resp.raise_for_status()
        text = resp.json().get("response", "").strip()
        if text:
            return text
        return None
    except Exception as exc:
        logger.warning("LLM gateway failed, falling back to local Ollama: %s", exc)
        return None


def _detect_ollama_model() -> str:
    """Auto-detect an available Ollama model if the configured one is missing."""
    base = OLLAMA_URL.rsplit("/api/", 1)[0] if "/api/" in OLLAMA_URL else "http://localhost:11434"
    try:
        r = requests.get(f"{base}/api/tags", timeout=5)
        if r.status_code != 200:
            return OLLAMA_MODEL
        models = [m.get("name", "") for m in r.json().get("models", [])]
        if not models:
            return OLLAMA_MODEL
        # Check if configured model is available (exact or prefix match)
        for m in models:
            if m == OLLAMA_MODEL or m.startswith(OLLAMA_MODEL.split(":")[0]):
                return OLLAMA_MODEL  # Configured model found
        # Configured model not found — pick best available (skip embedding models)
        embed_names = ("nomic", "embed", "bge", "all-minilm")
        gen_models = [m for m in models if not any(e in m.lower() for e in embed_names)]
        if gen_models:
            logger.warning("Configured model '%s' not found. Using '%s' instead.", OLLAMA_MODEL, gen_models[0])
            return gen_models[0]
        return OLLAMA_MODEL
    except Exception:
        return OLLAMA_MODEL


# Auto-detect model at startup (prevents 404 if client pulled a different model)
OLLAMA_MODEL = _detect_ollama_model()

# Support email for error reporting
SUPPORT_EMAIL = os.getenv("SUPPORT_EMAIL", "biznodesupport@1bz.biz")

# Track reported errors to avoid spamming support (error_key -> timestamp)
_reported_errors: dict[str, float] = {}
_ERROR_REPORT_COOLDOWN = 3600  # 1 hour between same error reports


def _report_error_to_support(error_type: str, error_detail: str, context: str = ""):
    """Send critical errors to support email for tracking. Rate-limited per error type."""
    import time as _time
    now = _time.time()
    if error_type in _reported_errors and (now - _reported_errors[error_type]) < _ERROR_REPORT_COOLDOWN:
        return  # Skip — already reported recently

    smtp_host = os.getenv("SMTP_HOST", "")
    smtp_user = os.getenv("SMTP_USER", "")
    smtp_pass = os.getenv("SMTP_PASSWORD", "")
    if not smtp_host or not smtp_user or not smtp_pass:
        return  # SMTP not configured on this node

    node_name = os.getenv("BOT_NAME", "BizNode")
    node_tier = os.getenv("NODE_TIER", "basic")
    owner_email = os.getenv("OWNER_EMAIL", "unknown")

    subject = f"[BizNode Error] {node_name} ({node_tier}) — {error_type}"
    body = (
        f"Node: {node_name} ({node_tier})\n"
        f"Owner: {owner_email}\n"
        f"Error: {error_type}\n"
        f"Detail: {error_detail[:500]}\n"
    )
    if context:
        body += f"Context: {context}\n"
    body += f"\nTime: {_time.strftime('%Y-%m-%d %H:%M:%S UTC', _time.gmtime())}\n"
    body += "\n--- BizNode Bot Error Reporter ---"

    try:
        import smtplib as _smtp
        from email.mime.text import MIMEText as _MIMEText
        msg = _MIMEText(body, "plain")
        msg["From"] = smtp_user
        msg["To"] = SUPPORT_EMAIL
        msg["Subject"] = subject
        server = _smtp.SMTP(smtp_host, int(os.getenv("SMTP_PORT", "587")), timeout=10)
        server.starttls()
        server.login(smtp_user, smtp_pass)
        server.sendmail(smtp_user, [SUPPORT_EMAIL], msg.as_string())
        server.quit()
        _reported_errors[error_type] = now
        logger.info("Error reported to support: %s", error_type)
    except Exception as exc:
        logger.debug("Failed to report error to support: %s", exc)


# Owner Telegram ID — this user bypasses onboarding and gets admin access
_owner_id_raw = os.getenv("OWNER_TELEGRAM_ID", "").strip()
OWNER_ID: int | None = int(_owner_id_raw) if _owner_id_raw.isdigit() else None

# Telegram group monitoring
_group_id_raw  = os.getenv("TELEGRAM_GROUP_ID", "").strip()
TELEGRAM_GROUP_ID: int | None = int(_group_id_raw) if _group_id_raw.lstrip("-").isdigit() else None
TELEGRAM_GROUP_LINK = os.getenv("TELEGRAM_GROUP_LINK", "")

QDRANT_COLLECTION = "biznode_memory"
EMBED_DIM = int(os.getenv("EMBEDDING_SIZE", "768"))

# BizNode client registration API
CLIENT_REGISTER_URL = os.getenv("CLIENT_REGISTER_URL", "https://biznode.1bz.biz/api_client_register.php?action=register")


def _register_node_with_server():
    """Register this node with the BizNode server on first startup.

    Sends bot details (chat_id, bot name, email) so admin can track
    client installations and add them to the Telegram group.
    """
    owner_email = os.getenv("OWNER_EMAIL", "")
    if not owner_email:
        return  # No email configured yet

    # Read version file
    version = ""
    version_file = Path(_PROJECT_ROOT) / "VERSION"
    if version_file.exists():
        version = version_file.read_text(encoding="utf-8").strip()

    import platform
    payload = {
        "owner_email":       owner_email,
        "bot_name":          os.getenv("BOT_NAME", "BizNode"),
        "bot_username":      os.getenv("TELEGRAM_BOT_USERNAME", ""),
        "telegram_chat_id":  os.getenv("TELEGRAM_GROUP_ID", ""),
        "owner_telegram_id": os.getenv("OWNER_TELEGRAM_ID", ""),
        "node_tier":         os.getenv("NODE_TIER", "basic"),
        "business_name":     os.getenv("BUSINESS_NAME", ""),
        "ollama_model":      OLLAMA_MODEL,
        "db_type":           os.getenv("DB_TYPE", "sqlite"),
        "os_info":           f"{platform.system()} {platform.release()}",
        "version":           version,
    }
    try:
        r = requests.post(CLIENT_REGISTER_URL, json=payload, timeout=10)
        if r.status_code == 200 and r.json().get("ok"):
            logger.info("Node registered with BizNode server (%s)", owner_email)
        else:
            logger.debug("Node registration response: %s", r.text[:200])
    except Exception as exc:
        logger.debug("Node registration failed (non-critical): %s", exc)


# Register node on bot startup (runs once, non-blocking)
try:
    _register_node_with_server()
except Exception:
    pass

# ---------------------------------------------------------------------------
# Onboarding steps
# ---------------------------------------------------------------------------
STEP_NEW               = "new"
STEP_AWAITING_NAME     = "awaiting_name"
STEP_AWAITING_BUSINESS = "awaiting_business"
STEP_AWAITING_EMAIL    = "awaiting_email"
STEP_AWAITING_OTP      = "awaiting_otp"
STEP_VERIFIED          = "verified"

# ---------------------------------------------------------------------------
# Person types — every interaction is classified
# ---------------------------------------------------------------------------
PERSON_OWNER            = "owner"
PERSON_CUSTOMER         = "customer"
PERSON_BOT              = "bot"
PERSON_SERVICE_PROVIDER = "service_provider"
PERSON_INVESTOR         = "investor"
PERSON_VISITOR          = "visitor"
PERSON_GUEST            = "guest"
PERSON_GOVERNMENT       = "government"
PERSON_FINANCIER        = "financier"

VALID_PERSON_TYPES = {
    PERSON_OWNER, PERSON_CUSTOMER, PERSON_BOT, PERSON_SERVICE_PROVIDER,
    PERSON_INVESTOR, PERSON_VISITOR, PERSON_GUEST, PERSON_GOVERNMENT,
    PERSON_FINANCIER,
}

# Map display labels for /role command
PERSON_TYPE_LABELS = {
    PERSON_CUSTOMER:         "Customer",
    PERSON_SERVICE_PROVIDER: "Service Provider",
    PERSON_INVESTOR:         "Investor",
    PERSON_VISITOR:          "Visitor / Just Browsing",
    PERSON_GUEST:            "Guest",
    PERSON_GOVERNMENT:       "Government / Regulatory",
    PERSON_FINANCIER:        "Financier / Funding",
}

# ---------------------------------------------------------------------------
# /help intake state machine (in-memory, keyed by user_id)
# ---------------------------------------------------------------------------
# Each entry: {"step": int (1-4), "business_type": str, "services": str,
#              "target_audience": str, "challenges": str}
_help_intake: dict[int, dict] = {}

# /setup wizard state machine (in-memory, keyed by user_id)
# ---------------------------------------------------------------------------
_setup_wizard: dict[int, dict] = {}

SETUP_INDUSTRIES = [
    ("general", "General Business"),
    ("legal", "Legal / Law Firm"),
    ("medical", "Medical / Healthcare"),
    ("realestate", "Real Estate"),
    ("finance", "Finance / Trading"),
    ("education", "Education"),
    ("retail", "Retail / E-Commerce"),
    ("consulting", "Consulting"),
]

HELP_STEP_QUESTIONS = {
    1: "What type of business do you run?",
    2: "What services or products do you offer?",
    3: "Who is your target audience?",
    4: "What are your main business challenges?",
}

# ---------------------------------------------------------------------------
# Rate limiting (per user)
# ---------------------------------------------------------------------------
# Short-burst limit (anti-spam)
BURST_MAX_REQUESTS   = 5
BURST_WINDOW_SECONDS = 30
_burst_times: dict[int, list[float]] = defaultdict(list)

# Long rolling window limit (customer fairness)
LONG_MAX_REQUESTS   = 10
LONG_WINDOW_SECONDS = 21600  # 6 hours
_long_times: dict[int, list[float]] = defaultdict(list)


def is_rate_limited(user_id: int) -> tuple[bool, str]:
    """Return (limited, reason). Owner is never limited."""
    now = time.time()

    # Short-burst check
    _burst_times[user_id] = [t for t in _burst_times[user_id] if now - t < BURST_WINDOW_SECONDS]
    if len(_burst_times[user_id]) >= BURST_MAX_REQUESTS:
        return True, "burst"

    # Long-window check
    _long_times[user_id] = [t for t in _long_times[user_id] if now - t < LONG_WINDOW_SECONDS]
    if len(_long_times[user_id]) >= LONG_MAX_REQUESTS:
        return True, "long"

    _burst_times[user_id].append(now)
    _long_times[user_id].append(now)
    return False, ""


def is_owner(user_id: int) -> bool:
    return OWNER_ID is not None and user_id == OWNER_ID


def _time_greeting() -> str:
    """Return time-of-day greeting based on local clock."""
    from datetime import datetime
    h = datetime.now().hour
    if 5 <= h < 12:
        return "Good morning"
    elif 12 <= h < 17:
        return "Good afternoon"
    elif 17 <= h < 21:
        return "Good evening"
    else:
        return "Good night"


def _last_topic(user_id: int) -> str:
    """Return the user's last question (for context recall), or empty string."""
    try:
        history = get_history(user_id)
        user_msgs = [h for h in history if h.get("role") == "customer"]
        if user_msgs:
            return user_msgs[-1].get("message", "")[:100]
    except Exception:
        pass
    return ""


def _person_type_greeting(person_type: str, name: str) -> str:
    """Return a warm, role-specific greeting line."""
    greetings = {
        PERSON_OWNER:            f"Welcome back, Boss {name}!",
        PERSON_CUSTOMER:         f"Welcome back, {name}! Great to see you again.",
        PERSON_BOT:              f"Hello {name} (bot). Ready to exchange data.",
        PERSON_SERVICE_PROVIDER: f"Hello {name}! Thanks for partnering with us.",
        PERSON_INVESTOR:         f"Welcome, {name}. Here's your business pulse.",
        PERSON_VISITOR:          f"Hi {name}! Welcome — let me show you what we offer.",
        PERSON_GUEST:            f"Hello {name}! Feel free to explore our services.",
        PERSON_GOVERNMENT:       f"Welcome, {name}. How can we assist with compliance?",
        PERSON_FINANCIER:        f"Welcome, {name}. Let's talk growth and opportunity.",
    }
    return greetings.get(person_type, f"Hi {name}! Welcome.")


# ---------------------------------------------------------------------------
# Dashboard API helpers
# ---------------------------------------------------------------------------

def _dash_get(path: str, **params) -> dict | list:
    try:
        r = requests.get(
            f"{DASHBOARD_URL}{path}",
            params=params or None,
            headers=DASH_HEADERS,
            timeout=5,
        )
        if r.ok:
            return r.json()
        logger.warning("DASH GET %s → %s: %s", path, r.status_code, r.text[:80])
    except Exception as exc:
        logger.warning("DASH GET %s failed: %s", path, exc)
    return {}


def _dash_post(path: str, payload: dict) -> dict:
    try:
        r = requests.post(
            f"{DASHBOARD_URL}{path}",
            json=payload,
            headers=DASH_HEADERS,
            timeout=5,
        )
        if r.ok:
            return r.json()
        logger.warning("DASH POST %s → %s: %s", path, r.status_code, r.text[:80])
    except Exception as exc:
        logger.warning("DASH POST %s failed: %s", path, exc)
    return {}


def _dash_delete(path: str) -> None:
    try:
        requests.delete(f"{DASHBOARD_URL}{path}", headers=DASH_HEADERS, timeout=5)
    except Exception:
        pass


# ── Bot user state ────────────────────────────────────────────────────────────

def get_bot_user(telegram_id: int) -> dict:
    result = _dash_get(f"/api/bot-user/{telegram_id}")
    return result if isinstance(result, dict) else {}


def update_bot_user(telegram_id: int, **fields) -> None:
    result = _dash_post(f"/api/bot-user/{telegram_id}/update", fields)
    if not result.get("ok"):
        logger.warning("update_bot_user failed for %s: %s", telegram_id, result)


def _is_smtp_configured() -> bool:
    """Check if SMTP is configured (via dashboard settings test)."""
    result = _dash_get("/api/settings")
    if not result:
        return False
    values = result.get("values", {})
    smtp_user = values.get("SMTP_USER", "")
    smtp_pw = values.get("SMTP_PASSWORD", "")
    # Masked values mean it's configured
    return bool(smtp_user and smtp_pw)


def send_otp(telegram_id: int, email: str, name: str) -> str:
    """Send OTP email. Returns 'ok', 'smtp_not_configured', or 'error'."""
    result = _dash_post("/api/otp/send", {
        "telegram_id": str(telegram_id), "email": email, "name": name,
    })
    if result.get("ok", False):
        return "ok"
    error = result.get("error", "unknown")
    logger.error("OTP send failed for %s: %s", email, error)
    if "SMTP not configured" in error or "not configured" in error.lower():
        return "smtp_not_configured"
    _report_error_to_support("otp_email_failed", f"OTP email to {email} failed: {error}")
    return "error"


def verify_otp(telegram_id: int, code: str) -> tuple[bool, str]:
    result = _dash_post("/api/otp/verify", {
        "telegram_id": str(telegram_id), "code": code.strip(),
    })
    return result.get("verified", False), result.get("message", "Verification failed")


def create_lead(name: str, business: str, email: str, telegram_id: int) -> None:
    _dash_post("/api/leads", {
        "name": name, "business": business, "email": email,
        "telegram_id": str(telegram_id), "source": "telegram_bot",
    })


def upsert_group_lead(name: str, telegram_id: int, username: str) -> None:
    """Silently capture a group member as a lead (source=telegram_group)."""
    _dash_post("/api/leads", {
        "name": name or username or f"Group member {telegram_id}",
        "business": "Group member",
        "telegram_id": str(telegram_id),
        "source": "telegram_group",
    })


# ── Other dashboard calls ─────────────────────────────────────────────────────

def is_blocked(telegram_id: int) -> bool:
    data = _dash_get(f"/api/customers/blocked/{telegram_id}")
    return isinstance(data, dict) and data.get("blocked", False)


def log_query(telegram_id: int, username: str, query: str, response: str) -> None:
    _dash_post("/api/query-log", {
        "telegram_id": str(telegram_id), "username": username or "",
        "query": query, "response": response,
    })


def get_history(telegram_id: int) -> list:
    data = _dash_get(f"/api/conversation/{telegram_id}", limit=10)
    return data if isinstance(data, list) else []


def add_message(telegram_id: int, role: str, message: str) -> None:
    _dash_post(f"/api/conversation/{telegram_id}", {"role": role, "message": message})


def clear_history(telegram_id: int) -> None:
    _dash_delete(f"/api/conversation/{telegram_id}")


def get_profile() -> dict:
    data = _dash_get("/api/business-profile")
    return data if isinstance(data, dict) else {}


def get_menu_items() -> list:
    data = _dash_get("/api/menus", active_only=True)
    return data if isinstance(data, list) else []


def trigger_workflow(telegram_id: int, trigger_command: str) -> str | None:
    result = _dash_post(f"/api/workflow-state/{telegram_id}/start",
                        {"trigger_command": trigger_command})
    return result.get("prompt") if isinstance(result, dict) else None


def advance_workflow(telegram_id: int, user_input: str) -> str | None:
    result = _dash_post(f"/api/workflow-state/{telegram_id}/advance",
                        {"user_input": user_input})
    return result.get("response") if isinstance(result, dict) else None


def get_recent_leads(limit: int = 5) -> list:
    data = _dash_get("/api/leads", limit=limit)
    return data if isinstance(data, list) else []


def get_recent_queries(limit: int = 5) -> list:
    data = _dash_get("/api/query-logs", limit=limit)
    return data if isinstance(data, list) else []


def get_report() -> dict:
    data = _dash_get("/api/bot/report")
    return data if isinstance(data, dict) else {}


# ---------------------------------------------------------------------------
# Qdrant / Embedding / LLM
# ---------------------------------------------------------------------------

def ensure_qdrant_collection() -> None:
    try:
        r = requests.get(f"{QDRANT_URL}/collections/{QDRANT_COLLECTION}", timeout=5)
        if r.status_code == 404:
            requests.put(
                f"{QDRANT_URL}/collections/{QDRANT_COLLECTION}",
                json={"vectors": {"size": EMBED_DIM, "distance": "Cosine"}},
                timeout=10,
            )
            logger.info("Created Qdrant collection '%s'", QDRANT_COLLECTION)
    except Exception as exc:
        logger.warning("Qdrant collection check failed: %s", exc)


def embed_text(text: str) -> list[float] | None:
    try:
        r = requests.post(
            OLLAMA_EMBED_URL,
            json={"model": EMBED_MODEL, "prompt": text},
            timeout=60,
        )
        r.raise_for_status()
        return r.json().get("embedding")
    except Exception as exc:
        logger.warning("Embedding failed: %s", exc)
        return None


def _search_single_collection(collection: str, vector: list[float],
                               top_k: int = 5) -> list[tuple[float, str]]:
    """Search one Qdrant collection, return list of (score, snippet)."""
    try:
        r = requests.post(
            f"{QDRANT_URL}/collections/{collection}/points/search",
            json={"vector": vector, "limit": top_k, "with_payload": True},
            timeout=10,
        )
        r.raise_for_status()
        results = []
        for hit in r.json().get("result", []):
            score   = hit.get("score", 0.0)
            payload = hit.get("payload", {})
            text    = (payload.get("content") or payload.get("text") or "").strip()
            title   = payload.get("title", "")
            if text:
                snippet = (f"[{title}]\n" if title else "") + text[:1200]
                results.append((score, snippet))
        return results
    except Exception as exc:
        logger.warning("Qdrant search on %s failed: %s", collection, exc)
        return []


def search_memory(vector: list[float], top_k: int = 5) -> list[str]:
    """Search both biznode_memory and biznode_support, merge by score."""
    memory_hits  = _search_single_collection(QDRANT_COLLECTION, vector, top_k)
    support_hits = _search_single_collection("biznode_support", vector, top_k)
    combined = memory_hits + support_hits
    combined.sort(key=lambda x: x[0], reverse=True)
    return [snippet for _, snippet in combined[:top_k]]


def store_in_memory(text: str, vector: list[float], user_id: int) -> None:
    try:
        import hashlib, struct
        h = hashlib.sha256(f"{user_id}:{text}".encode()).digest()
        point_id = struct.unpack(">Q", h[:8])[0]
        requests.put(
            f"{QDRANT_URL}/collections/{QDRANT_COLLECTION}/points",
            json={"points": [{"id": point_id, "vector": vector,
                               "payload": {"text": text, "user_id": user_id}}]},
            timeout=10,
        )
    except Exception as exc:
        logger.warning("Qdrant store failed: %s", exc)


_INSTRUCTION_CACHE: str = ""


def _load_instruction() -> str:
    global _INSTRUCTION_CACHE
    try:
        data = _dash_get("/api/instruction")
        _INSTRUCTION_CACHE = data.get("content", "") if isinstance(data, dict) else ""
    except Exception:
        try:
            with open("instruction.md", encoding="utf-8") as f:
                _INSTRUCTION_CACHE = f.read()
        except Exception:
            _INSTRUCTION_CACHE = ""
    return _INSTRUCTION_CACHE


_SYSTEM_PROMPT = (
    "You are a friendly, professional AI business assistant for 1BZ DZIT DAO LLC — the 1BZ Node Network. "
    "You represent BizNode and the entire 1BZ ecosystem of decentralized AI business nodes. "
    "Answer questions using the Business instructions and Knowledge base provided in the prompt. "
    "If the answer is in the Business instructions or Knowledge base, answer from there directly. "
    "Be conversational, warm, and helpful — not robotic. Use the customer's name if available. "
    "For greetings like 'hi', 'hello', respond warmly and offer to help. "
    "For 'thank you', respond graciously. For casual chat, be friendly but steer toward business topics. "
    "NEVER say you don't have information about this business or its services — you ARE this network's assistant. "
    "You know about ALL 1BZ nodes: BizNode, BizChannel, DZIT, SmartPDF, CopyGuard, IPVault, Fax, UBC, DX, FV. "
    "If you genuinely don't know something, offer to connect them with the business owner."
)

_OWNER_SYSTEM_PROMPT = (
    "You are an intelligent AI sales assistant for the BUSINESS OWNER of 1BZ DZIT DAO LLC. "
    "You help the owner SELL BizNode™ — an AI-powered business automation platform. "
    "Respond naturally like a smart executive assistant — not a command-line tool. "
    "Answer questions directly, factually, and conversationally. "
    "You have access to the business knowledge base AND live CRM data — use them to give accurate answers. "
    "If the owner's question refers to something discussed before, acknowledge the context naturally. "
    "When live CRM data is provided below, use those real numbers — do NOT say you lack access to data. "
    "\n\nYou can TAKE ACTIONS on behalf of the owner:\n"
    "• Send emails — the owner can say 'email X about Y' and you compose + send it\n"
    "• Block/unblock users by Telegram ID\n"
    "• Show leads, reports, status, queries, profile\n"
    "• Update settings (email, phone) with OTP verification\n"
    "\nWhen the owner asks you to do something, DO IT or explain how. "
    "Do NOT just show command formats — be proactive and helpful. "
    "If the owner says 'mail ceo@1bz.biz about project update', compose and send a professional email. "
    "If the owner asks 'how are my leads today', give the actual numbers from CRM data. "
    "Think and respond like a capable human assistant, not a bot reading a manual."
    "\n\n--- EMAIL WRITING RULES ---\n"
    "When asked to write/send an email about BizNode or 'explaining about biznode', ALWAYS write a SALES email.\n"
    "NEVER write a partnership, collaboration, or joint venture email unless the owner explicitly says 'partnership'.\n"
    "The DEFAULT intent for 'explain about biznode' or 'about biznode' = SELL the product.\n"
    "Sales emails MUST include:\n"
    "1. A compelling subject line focused on business value\n"
    "2. Opening that addresses the recipient's business challenges (time, manual work, missed leads)\n"
    "3. What BizNode does: AI-powered business automation — runs 24/7, handles leads, CRM, customer support, email automation, scheduling, analytics\n"
    "4. Key features by tier:\n"
    "   - Basic ($200): AI chatbot, lead capture, CRM, support knowledge base, email automation, scheduling, analytics, local backup\n"
    "   - Pro ($499): Everything in Basic PLUS BizChannel ad marketplace, AI agent employment, agent monitoring, "
    "Stripe/crypto payments, affiliate system, cloud backup, API access, priority support, multi-bot (up to 5)\n"
    "5. ROI angle: replaces manual work, captures leads 24/7, never misses a customer query, runs locally (no cloud dependency)\n"
    "6. 10-day free trial available\n"
    "7. Call-to-action: Visit https://biznode.1bz.biz or contact @biznode_bot on Telegram\n"
    "8. Professional sign-off from the business owner (1BZ DZIT DAO LLC), NOT generic 'BizNode Team'\n"
    "Tone: persuasive but professional. Focus on problems solved and ROI, not technical jargon.\n"
)


def _get_owner_crm_snapshot() -> str:
    """Fetch a real-time CRM snapshot to inject into owner LLM prompts."""
    parts = []
    try:
        report = get_report()
        if report and "error" not in report:
            parts.append(
                f"LIVE CRM DATA (as of now):\n"
                f"- Queries today: {report.get('queries_today', 0)}\n"
                f"- Unique users today: {report.get('unique_users_today', 0)}\n"
                f"- New leads today: {report.get('new_leads_today', 0)}\n"
                f"- Total leads: {report.get('total_leads', 0)}\n"
                f"- Verified customers: {report.get('verified_customers', 0)}\n"
                f"- Total bot users: {report.get('total_bot_users', 0)}"
            )
    except Exception:
        pass
    try:
        leads = get_recent_leads(5)
        if leads:
            lead_lines = []
            for l in leads:
                name = l.get("name") or l.get("telegram_id") or "Unknown"
                email = l.get("email") or l.get("contact_info") or ""
                status = l.get("status") or "new"
                masked = _mask_email(email) if email else ""
                lead_lines.append(f"  - {name} ({masked}) [{status}]")
            parts.append("Recent leads:\n" + "\n".join(lead_lines))
    except Exception:
        pass
    try:
        profile = get_profile()
        if profile:
            parts.append(
                f"Business profile: {profile.get('business_name', 'N/A')}"
            )
    except Exception:
        pass
    return "\n\n".join(parts)


def query_llm(prompt: str, context_snippets: list[str] | None = None,
              history: list[dict] | None = None,
              owner_mode: bool = False) -> str:
    system = _OWNER_SYSTEM_PROMPT if owner_mode else _SYSTEM_PROMPT
    parts = []
    if owner_mode:
        crm = _get_owner_crm_snapshot()
        if crm:
            parts.append(crm)
    elif _INSTRUCTION_CACHE:
        parts.append(f"Business instructions:\n{_INSTRUCTION_CACHE}")
    if context_snippets:
        filtered = [s for s in context_snippets if s and len(s) > 10]
        if filtered:
            parts.append("Knowledge base:\n\n" + "\n\n---\n\n".join(filtered))
    if history and not owner_mode:
        lines = "\n".join(
            f"{'Customer' if h['role'] == 'customer' else 'Assistant'}: {h['message']}"
            for h in history
        )
        parts.append(f"Conversation history:\n{lines}")
    parts.append(f"Question: {prompt}\n\nAnswer:")
    full_prompt = "\n\n".join(parts)

    # Try LLM gateway first (better model, centralized)
    gateway_result = _call_llm_gateway(full_prompt, system)
    if gateway_result:
        return gateway_result

    # Fallback to local Ollama
    try:
        r = requests.post(
            OLLAMA_URL,
            json={"model": OLLAMA_MODEL, "prompt": full_prompt,
                  "system": system, "stream": False},
            timeout=300,
        )
        r.raise_for_status()
        return r.json().get("response", "No response from AI.").strip()
    except requests.exceptions.Timeout:
        _report_error_to_support("ollama_timeout", "LLM query timed out after 5 minutes", f"model={OLLAMA_MODEL}")
        return "AI took too long (5 min). The model may be loading. Please try again — it should be faster now."
    except requests.exceptions.ConnectionError:
        _report_error_to_support("ollama_connection_error", f"Cannot connect to {OLLAMA_URL}", f"model={OLLAMA_MODEL}")
        return "AI service is offline right now. Please try again later."
    except Exception as exc:
        logger.error("LLM query failed: %s", exc)
        _report_error_to_support("llm_query_failed", str(exc), f"model={OLLAMA_MODEL}")
        return f"Error talking to AI: {exc}"


# ---------------------------------------------------------------------------
# Tier 2: LLM Tool Calling (owner only)
# ---------------------------------------------------------------------------
# Ollama /api/chat with tool definitions — LLM decides what action to take.

OLLAMA_CHAT_URL = OLLAMA_URL.replace("/api/generate", "/api/chat")

_OWNER_TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "send_email",
            "description": "Compose and send a professional sales email on behalf of the business owner. Default to sales/promotional tone when topic is about BizNode.",
            "parameters": {
                "type": "object",
                "properties": {
                    "to": {"type": "string", "description": "Recipient email address"},
                    "topic": {"type": "string", "description": "What the email should be about"},
                },
                "required": ["to", "topic"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_leads",
            "description": "Get recent customer leads, prospects, and sign-ups",
            "parameters": {
                "type": "object",
                "properties": {
                    "count": {"type": "integer", "description": "Number of leads to return (default 10)"},
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_report",
            "description": "Get today's business report: queries, leads, users, verified customers",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_status",
            "description": "Check system health: Ollama, Qdrant, Dashboard status",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_knowledge_base",
            "description": "Search the business knowledge base for information about BizNode features, pricing, setup, or any business topic",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"},
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_recent_queries",
            "description": "Get recent customer questions and chat queries",
            "parameters": {
                "type": "object",
                "properties": {
                    "count": {"type": "integer", "description": "Number of queries (default 10)"},
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "generate_vault_report",
            "description": "Generate a DataVault report using a plugin (e.g., sales report, competitive intel, ICP). Available commands: salesreport, competitors, icp, outreach, objections, proposal, growth, listings",
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {"type": "string", "description": "Plugin command name (salesreport, competitors, icp, outreach, objections, proposal, growth, listings)"},
                    "args": {"type": "string", "description": "Command arguments like URL or description"},
                },
                "required": ["command"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "browse_vault",
            "description": "Browse or search DataVault reports. Shows available reports with prices.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query (optional, empty shows latest reports)"},
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_email_log",
            "description": "Get sent/received email history. Use to recall what emails were sent, to whom, and when.",
            "parameters": {
                "type": "object",
                "properties": {
                    "direction": {"type": "string", "enum": ["sent", "received"], "description": "Filter by sent or received (optional)"},
                    "count": {"type": "integer", "description": "Number of emails to return (default 10)"},
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_conversation_history",
            "description": "Recall past conversations with a specific customer by their Telegram ID. Shows what they asked and what the bot replied.",
            "parameters": {
                "type": "object",
                "properties": {
                    "telegram_id": {"type": "string", "description": "Customer's Telegram ID"},
                    "count": {"type": "integer", "description": "Number of messages to return (default 10)"},
                },
                "required": ["telegram_id"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_scheduled_tasks",
            "description": "List all scheduled tasks (cron jobs, reminders, automations). Shows task name, schedule, and status.",
            "parameters": {
                "type": "object",
                "properties": {
                    "status": {"type": "string", "enum": ["active", "paused", "completed"], "description": "Filter by status (optional)"},
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_blocked_customers",
            "description": "List all currently blocked customers with reason and date blocked.",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_bot_users",
            "description": "Get list of bot users — people who have interacted with the bot. Shows name, Telegram ID, verified status, and last seen.",
            "parameters": {
                "type": "object",
                "properties": {
                    "count": {"type": "integer", "description": "Number of users to return (default 20)"},
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_transactions",
            "description": "Get payment transaction history. Shows amount, status, customer, and date.",
            "parameters": {
                "type": "object",
                "properties": {
                    "status": {"type": "string", "enum": ["pending", "completed", "failed", "refunded"], "description": "Filter by status (optional)"},
                    "count": {"type": "integer", "description": "Number of transactions to return (default 20)"},
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_license_purchases",
            "description": "Get license purchase history. Shows buyer, license key, tier, payment status, and date.",
            "parameters": {
                "type": "object",
                "properties": {
                    "status": {"type": "string", "description": "Filter by status (optional)"},
                    "count": {"type": "integer", "description": "Number of purchases to return (default 20)"},
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_backup_history",
            "description": "Get backup history — when backups ran, whether they succeeded, size, and destination.",
            "parameters": {
                "type": "object",
                "properties": {
                    "count": {"type": "integer", "description": "Number of entries to return (default 10)"},
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_client_files",
            "description": "Get client file records — business profiles, contact details, and status of onboarded clients.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search by name, email, or business (optional)"},
                    "count": {"type": "integer", "description": "Number of clients to return (default 20)"},
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_owner_actions",
            "description": "Get history of owner commands and actions — what instructions were given and their results.",
            "parameters": {
                "type": "object",
                "properties": {
                    "count": {"type": "integer", "description": "Number of actions to return (default 20)"},
                },
                "required": [],
            },
        },
    },
]


def _execute_tool(name: str, args: dict) -> str:
    """Execute a tool call and return the result as a string for the LLM."""
    try:
        if name == "send_email":
            to_email = _normalize_email(args.get("to", ""))
            topic = args.get("topic", "general inquiry")
            # Compose email body using LLM
            compose_prompt = (
                f"Write a professional SALES email body about: {topic}\n"
                f"Recipient: {to_email}\n"
                f"From: The business owner of 1BZ DZIT DAO LLC.\n"
                f"IMPORTANT: If the topic is about BizNode or explaining BizNode, write a SALES email that:\n"
                f"- Highlights what BizNode does (AI business automation: chatbot, CRM, lead capture, email automation, analytics)\n"
                f"- Mentions pricing: Basic $200, Pro $499, 10-day free trial\n"
                f"- Focuses on ROI and problems solved (24/7 automation, never miss leads, no cloud dependency)\n"
                f"- Includes CTA: visit https://biznode.1bz.biz or contact @biznode_bot on Telegram\n"
                f"- Signs off as the business owner, NOT as 'BizNode Team'\n"
                f"- NEVER write a partnership or collaboration email unless explicitly asked\n"
                f"Be persuasive, professional, and concise. Just the body text, no subject line."
            )
            body = query_llm(compose_prompt, owner_mode=True)
            subject = topic.title() if len(topic) < 80 else topic[:77] + "..."
            from services.email_service import EmailService
            svc = EmailService()
            svc.send_email(to_email, subject, body)
            return f"Email sent successfully to {to_email}.\nSubject: {subject}\nBody:\n{body[:500]}"

        elif name == "get_leads":
            count = args.get("count", 10)
            leads = get_recent_leads(count)
            report = get_report()
            if leads:
                lines = []
                for i, l in enumerate(leads, 1):
                    n = l.get("name") or l.get("telegram_id") or "Unknown"
                    e = l.get("email") or l.get("contact_info") or ""
                    s = l.get("status") or "new"
                    lines.append(f"{i}. {n} — {_mask_email(e) if e else 'no email'} [{s}]")
                return (
                    f"Today: {report.get('new_leads_today', 0)} new leads. "
                    f"Total: {report.get('total_leads', 0)}. "
                    f"Bot users: {report.get('total_bot_users', 0)}.\n\n"
                    f"Recent leads:\n" + "\n".join(lines)
                )
            return f"No leads yet. Total bot users: {report.get('total_bot_users', 0)}"

        elif name == "get_report":
            report = get_report()
            return (
                f"Queries today: {report.get('queries_today', 0)}\n"
                f"Unique users today: {report.get('unique_users_today', 0)}\n"
                f"New leads today: {report.get('new_leads_today', 0)}\n"
                f"Total leads: {report.get('total_leads', 0)}\n"
                f"Verified customers: {report.get('verified_customers', 0)}\n"
                f"Total bot users: {report.get('total_bot_users', 0)}"
            )

        elif name == "get_status":
            try:
                r = requests.get(OLLAMA_URL.replace("/api/generate", ""), timeout=3)
                ollama_status = "running" if r.status_code < 500 else "error"
            except Exception:
                ollama_status = "offline"
            try:
                r = requests.get(f"{QDRANT_URL}/collections/{QDRANT_COLLECTION}", timeout=3)
                pts = r.json().get("result", {}).get("points_count", "?")
                qdrant_status = f"running ({pts} KB entries)"
            except Exception:
                qdrant_status = "offline"
            return (
                f"Ollama: {ollama_status} (model: {OLLAMA_MODEL})\n"
                f"Qdrant: {qdrant_status}\n"
                f"Dashboard: {DASHBOARD_URL}"
            )

        elif name == "search_knowledge_base":
            query = args.get("query", "")
            vector = embed_text(query)
            if vector:
                snippets = search_memory(vector)
                if snippets:
                    return "\n\n---\n\n".join(snippets[:3])
            return "No relevant information found in knowledge base."

        elif name == "get_recent_queries":
            count = args.get("count", 10)
            rows = get_recent_queries(count)
            if rows:
                lines = []
                for i, q in enumerate(rows, 1):
                    who = q.get("username") or q.get("telegram_id") or "?"
                    query = (q.get("query") or "")[:60]
                    lines.append(f"{i}. [{who}] {query}")
                return "Recent customer queries:\n" + "\n".join(lines)
            return "No customer queries yet."

        elif name == "generate_vault_report":
            command = args.get("command", "salesreport")
            cmd_args = args.get("args", "")
            try:
                from tools.admin.datavault.report_generator import generate_report
                result = generate_report(command=command, args=cmd_args, creator_node="admin")
                if result.get("success"):
                    return (
                        f"Report generated: {result['title']}\n"
                        f"Report ID: {result['report_id']}\n"
                        f"PDF: {result['pdf_path']}"
                    )
                return f"Generation failed: {result.get('message', 'Unknown error')}"
            except Exception as exc:
                return f"DataVault error: {exc}"

        elif name == "browse_vault":
            query = args.get("query", "")
            try:
                from tools.admin.datavault.datavault_service import search_reports, list_reports
                if query:
                    reports = search_reports(query, limit=5)
                else:
                    reports = list_reports(limit=5)
                if reports:
                    lines = []
                    for r in reports:
                        price_str = f"{r['price']} {r['currency']}" if r['price'] > 0 else "FREE"
                        lines.append(f"- {r['title']} [{price_str}] (ID: {r['report_id']})")
                    return "DataVault reports:\n" + "\n".join(lines)
                return "No reports found in the vault."
            except Exception as exc:
                return f"DataVault error: {exc}"

        elif name == "get_email_log":
            direction = args.get("direction")
            count = args.get("count", 10)
            try:
                from memory.database import get_email_log as _db_email_log
                rows = _db_email_log(direction=direction, limit=count)
                if rows:
                    lines = []
                    for e in rows:
                        d = e.get("direction", "?")
                        to = _mask_email(e.get("to_addr", "")) if e.get("to_addr") else ""
                        fr = _mask_email(e.get("from_addr", "")) if e.get("from_addr") else ""
                        subj = e.get("subject", "")[:60]
                        ts = e.get("created_at", "")
                        status = e.get("status", "ok")
                        addr = to if d == "sent" else fr
                        lines.append(f"[{d}] {addr} — {subj} ({ts}) [{status}]")
                    return f"Email log ({len(rows)} entries):\n" + "\n".join(lines)
                return "No emails found in the log."
            except Exception as exc:
                return f"Email log error: {exc}"

        elif name == "get_conversation_history":
            tid = args.get("telegram_id", "")
            count = args.get("count", 10)
            from memory.database import get_conversation_history as _db_conv
            rows = _db_conv(str(tid), limit=count)
            if rows:
                lines = []
                for c in rows:
                    role = c.get("role", "?")
                    msg = (c.get("content") or "")[:120]
                    ts = c.get("created_at", "")
                    lines.append(f"[{role}] {msg} ({ts})")
                return f"Conversation with {tid} ({len(rows)} msgs):\n" + "\n".join(lines)
            return f"No conversation history found for {tid}."

        elif name == "get_scheduled_tasks":
            status = args.get("status")
            from memory.database import get_scheduled_tasks as _db_tasks
            rows = _db_tasks(status=status)
            if rows:
                lines = []
                for t in rows:
                    lines.append(
                        f"- {t.get('task_name', '?')} [{t.get('status', '?')}] "
                        f"schedule: {t.get('schedule', '?')} | next: {t.get('next_run', '?')}"
                    )
                return f"Scheduled tasks ({len(rows)}):\n" + "\n".join(lines)
            return "No scheduled tasks found."

        elif name == "get_blocked_customers":
            from memory.database import get_blocked_customers as _db_blocked
            rows = _db_blocked()
            if rows:
                lines = []
                for b in rows:
                    lines.append(
                        f"- ID {b.get('telegram_id', '?')} — {b.get('reason', 'no reason')} "
                        f"(blocked: {b.get('created_at', '?')})"
                    )
                return f"Blocked customers ({len(rows)}):\n" + "\n".join(lines)
            return "No blocked customers."

        elif name == "get_bot_users":
            count = args.get("count", 20)
            from memory.database import get_connection
            conn = get_connection()
            rows = conn.execute(
                "SELECT telegram_id, name, username, email, business, step, verified, "
                "last_seen, created_at FROM bot_users ORDER BY last_seen DESC LIMIT ?",
                (count,)
            ).fetchall()
            conn.close()
            if rows:
                lines = []
                for u in [dict(r) for r in rows]:
                    display = u.get("name", "") or u.get("username", "") or "?"
                    status = "verified" if u.get("verified") else f"step:{u.get('step', '?')}"
                    email = u.get("email", "")
                    biz = u.get("business", "")
                    info = f"- {display} (ID: {u['telegram_id']}) [{status}]"
                    if email:
                        info += f" email: {_mask_email(email)}"
                    if biz:
                        info += f" biz: {biz}"
                    info += f" last: {u.get('last_seen', '?')}"
                    lines.append(info)
                return f"Bot users ({len(rows)}):\n" + "\n".join(lines)
            return "No bot users yet."

        elif name == "get_transactions":
            status = args.get("status")
            count = args.get("count", 20)
            from memory.database import get_all_transactions as _db_txn
            rows = _db_txn(status=status, limit=count)
            if rows:
                lines = []
                for t in rows:
                    lines.append(
                        f"- {t.get('reference_no', '?')} | {t.get('amount', 0)} {t.get('currency', '')} "
                        f"[{t.get('status', '?')}] {t.get('description', '')[:40]} ({t.get('created_at', '')})"
                    )
                return f"Transactions ({len(rows)}):\n" + "\n".join(lines)
            return "No transactions found."

        elif name == "get_license_purchases":
            status = args.get("status")
            count = args.get("count", 20)
            from memory.database import get_license_purchases as _db_lic
            rows = _db_lic(status=status, limit=count)
            if rows:
                lines = []
                for p in rows:
                    key = p.get("license_key", "pending")
                    lines.append(
                        f"- {p.get('reference_no', '?')} | {p.get('buyer_name', '?')} | "
                        f"{p.get('tier', '?')} | key: {key} [{p.get('status', '?')}] ({p.get('created_at', '')})"
                    )
                return f"License purchases ({len(rows)}):\n" + "\n".join(lines)
            return "No license purchases found."

        elif name == "get_backup_history":
            count = args.get("count", 10)
            from memory.database import get_backup_history as _db_bak
            rows = _db_bak(limit=count)
            if rows:
                lines = []
                for b in rows:
                    lines.append(
                        f"- {b.get('created_at', '?')} | {b.get('provider', '?')} "
                        f"[{b.get('status', '?')}] size: {b.get('size_bytes', 0)} bytes"
                    )
                return f"Backup history ({len(rows)}):\n" + "\n".join(lines)
            return "No backup history found."

        elif name == "get_client_files":
            query = args.get("query", "")
            count = args.get("count", 20)
            from memory.database import search_client_files, get_all_client_files
            rows = search_client_files(query) if query else get_all_client_files(limit=count)
            if rows:
                lines = []
                for c in rows[:count]:
                    lines.append(
                        f"- {c.get('business_name', '?')} | {c.get('contact_name', '?')} "
                        f"| {_mask_email(c.get('email', '') or '')} [{c.get('status', '?')}]"
                    )
                return f"Client files ({len(rows)}):\n" + "\n".join(lines)
            return "No client files found."

        elif name == "get_owner_actions":
            count = args.get("count", 20)
            from memory.database import get_pending_actions
            from memory.database import get_connection
            conn = get_connection()
            rows = conn.execute(
                "SELECT action_type, description, status, result, created_at "
                "FROM owner_actions ORDER BY created_at DESC LIMIT ?",
                (count,)
            ).fetchall()
            conn.close()
            if rows:
                lines = []
                for a in [dict(r) for r in rows]:
                    lines.append(
                        f"- [{a.get('action_type', '?')}] {(a.get('description') or '')[:60]} "
                        f"→ {a.get('status', '?')} ({a.get('created_at', '')})"
                    )
                return f"Owner actions ({len(rows)}):\n" + "\n".join(lines)
            return "No owner action history found."

        else:
            return f"Unknown tool: {name}"

    except Exception as exc:
        logger.error("Tool %s failed: %s", name, exc, exc_info=True)
        return f"Tool error: {exc}"


def owner_tool_answer(user_id: int, text: str, username: str) -> str:
    """Tier 2: Send owner message to LLM with tool calling.

    The LLM decides whether to respond conversationally or call a tool.
    If a tool is called, execute it and feed the result back to the LLM
    for a natural language final response.
    """
    # Build context
    crm = _get_owner_crm_snapshot()
    system_msg = _OWNER_SYSTEM_PROMPT
    if crm:
        system_msg += f"\n\n--- LIVE CRM DATA ---\n{crm}"

    messages = [
        {"role": "system", "content": system_msg},
        {"role": "user", "content": text},
    ]

    try:
        # Step 1: Ask LLM (with tools)
        r = requests.post(
            OLLAMA_CHAT_URL,
            json={
                "model": OLLAMA_MODEL,
                "messages": messages,
                "tools": _OWNER_TOOLS,
                "stream": False,
            },
            timeout=120,
        )
        r.raise_for_status()
        data = r.json()
        msg = data.get("message", {})
        tool_calls = msg.get("tool_calls", [])

        # No tool call — return text response directly
        if not tool_calls:
            response = (msg.get("content") or "").strip()
            if not response:
                response = "I'm not sure how to help with that. Could you rephrase?"
            # Save conversation
            try:
                add_message(user_id, "customer", text)
                add_message(user_id, "assistant", response)
            except Exception:
                pass
            log_query(user_id, username, text, response)
            return response

        # Step 2: Execute tool(s)
        tool_results = []
        for tc in tool_calls:
            fn = tc.get("function", {})
            fn_name = fn.get("name", "")
            fn_args = fn.get("arguments", {})
            logger.info("TOOL_CALL name=%s args=%s", fn_name, fn_args)
            result = _execute_tool(fn_name, fn_args)
            tool_results.append({"name": fn_name, "result": result})

        # Step 3: Feed tool results back to LLM for final response
        messages.append(msg)  # assistant message with tool_calls
        for tr in tool_results:
            messages.append({
                "role": "tool",
                "content": tr["result"],
            })

        r2 = requests.post(
            OLLAMA_CHAT_URL,
            json={
                "model": OLLAMA_MODEL,
                "messages": messages,
                "stream": False,
            },
            timeout=120,
        )
        r2.raise_for_status()
        final = r2.json().get("message", {}).get("content", "").strip()
        if not final:
            # Fallback: show raw tool result
            final = "\n\n".join(
                f"[{tr['name']}]\n{tr['result']}" for tr in tool_results
            )

        # Save conversation
        try:
            add_message(user_id, "customer", text)
            add_message(user_id, "assistant", final)
        except Exception:
            pass
        log_query(user_id, username, text, final)
        return final

    except requests.exceptions.Timeout:
        return "AI took too long. The model may be loading — please try again."
    except requests.exceptions.ConnectionError:
        return "AI service is offline. Please try again later."
    except Exception as exc:
        logger.error("Owner tool calling failed: %s", exc, exc_info=True)
        # Fallback to regular RAG
        return query_llm(text, owner_mode=True)


def rag_answer(user_id: int, text: str, username: str, owner_mode: bool = False) -> str:
    """Full RAG pipeline: embed → search → LLM → log.

    owner_mode=True: uses owner system prompt, no vector store writes, no
    instruction.md persona. Conversation saved for context recall on /start.
    """
    history = [] if owner_mode else get_history(user_id)
    vector  = embed_text(text)
    snippets: list[str] = []
    if vector:
        snippets = search_memory(vector)
        if not owner_mode:
            store_in_memory(text, vector, user_id)
    response = query_llm(text, snippets, history=history, owner_mode=owner_mode)
    # Always save owner conversation so _last_topic() works on next /start
    try:
        add_message(user_id, "customer", text)
        add_message(user_id, "assistant", response)
    except Exception:
        pass
    log_query(user_id, username, text, response)
    return response


# ---------------------------------------------------------------------------
# UI helpers
# ---------------------------------------------------------------------------
_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
_EMAIL_EXTRACT_RE = re.compile(r"[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}")


def _valid_email(text: str) -> bool:
    return bool(_EMAIL_RE.match(text.strip()))


def _extract_email(text: str) -> str | None:
    """Extract email from text even if surrounded by other words."""
    text = text.strip()
    # Direct match first
    if _valid_email(text):
        return text
    # Try to find email embedded in text like "my email is abc@xyz.com"
    match = _EMAIL_EXTRACT_RE.search(text)
    return match.group(0) if match else None


def _main_menu(person_type: str = PERSON_VISITOR) -> InlineKeyboardMarkup:
    """Build menu based on person type. Combines DB menu items + built-in categories."""
    items    = get_menu_items()
    keyboard = []

    # DB-configured menu items (from dashboard Menu Builder)
    for item in items:
        if item.get("menu_type") in ("web_link", "bot_link"):
            keyboard.append([InlineKeyboardButton(item["menu_name"],
                                                   url=item.get("link", ""))])
        else:
            task_code = item.get("task_code") or item.get("menu_name")
            keyboard.append([InlineKeyboardButton(item["menu_name"],
                                                   callback_data=f"task:{task_code}")])

    # ── Built-in categories (always present) ──
    keyboard.append([InlineKeyboardButton("Our Services", callback_data="info:services")])
    keyboard.append([InlineKeyboardButton("Buy a License", callback_data="info:buy_license")])
    keyboard.append([InlineKeyboardButton("Partner Links", callback_data="info:affiliates")])

    # Role-specific menu items
    if person_type == PERSON_CUSTOMER:
        keyboard.append([InlineKeyboardButton("My Services", callback_data="info:my_services")])
        keyboard.append([InlineKeyboardButton("Request Price Quote", callback_data="info:price_quote")])
        keyboard.append([InlineKeyboardButton("Payment History", callback_data="pay:history")])
    elif person_type == PERSON_INVESTOR:
        keyboard.append([InlineKeyboardButton("Business Overview", callback_data="info:biz_overview")])
        keyboard.append([InlineKeyboardButton("Growth Metrics", callback_data="info:growth")])
    elif person_type == PERSON_SERVICE_PROVIDER:
        keyboard.append([InlineKeyboardButton("Partnership Info", callback_data="info:partnership")])
    elif person_type == PERSON_FINANCIER:
        keyboard.append([InlineKeyboardButton("Financial Overview", callback_data="info:finance")])

    # Common items for everyone
    keyboard.append([InlineKeyboardButton("Ask a Question", callback_data="task:__ask")])
    keyboard.append([InlineKeyboardButton("About BizNode AI", callback_data="info:about_biznode")])
    keyboard.append([InlineKeyboardButton("Change My Role", callback_data="info:change_role")])
    return InlineKeyboardMarkup(keyboard)


def _role_selection_menu() -> InlineKeyboardMarkup:
    """Inline keyboard for selecting person type."""
    keyboard = []
    for ptype, label in PERSON_TYPE_LABELS.items():
        keyboard.append([InlineKeyboardButton(label, callback_data=f"role:{ptype}")])
    return InlineKeyboardMarkup(keyboard)


def _owner_menu() -> InlineKeyboardMarkup:
    """Full owner menu organized by categories."""
    return InlineKeyboardMarkup([
        # ── CRM & Business ──
        [InlineKeyboardButton("My Profile",       callback_data="owner:profile")],
        [InlineKeyboardButton("Recent Leads",     callback_data="owner:leads")],
        [InlineKeyboardButton("Daily Report",     callback_data="owner:report")],
        [InlineKeyboardButton("Recent Queries",   callback_data="owner:queries")],
        # ── System ──
        [InlineKeyboardButton("Server Status",    callback_data="owner:status")],
        # ── Services & Payments ──
        [InlineKeyboardButton("Services & Pricing",  callback_data="owner:services")],
        [InlineKeyboardButton("License Purchases",   callback_data="owner:purchases")],
        [InlineKeyboardButton("Transactions",        callback_data="owner:transactions")],
        # ── Marketing ──
        [InlineKeyboardButton("Affiliate Links",     callback_data="owner:affiliates")],
        # ── Tools ──
        [InlineKeyboardButton("Send Email",          callback_data="owner:sendemail")],
        [InlineKeyboardButton("Knowledge Base",      callback_data="owner:kb")],
        [InlineKeyboardButton("Dashboard",           url="http://localhost:7777")],
        # ── Customer view ──
        [InlineKeyboardButton("Customer Menu Preview", callback_data="owner:custmenu")],
    ])


# ---------------------------------------------------------------------------
# Owner handlers
# ---------------------------------------------------------------------------

async def _owner_start(update: Update) -> None:
    user_id    = update.effective_user.id
    first_name = (update.effective_user.first_name or "").strip() or "Boss"

    # Contextual time greeting
    greeting   = _time_greeting()

    # Business profile
    profile  = get_profile()
    biz_name = (profile.get("business_name") or "your business").strip()
    persona  = (profile.get("ai_persona_name") or "BizNode").strip()

    # Activity report
    report   = get_report()
    q_today  = report.get("queries_today", 0)
    u_today  = report.get("unique_users_today", 0)
    l_today  = report.get("new_leads_today", 0)
    t_leads  = report.get("total_leads", 0)
    verified = report.get("verified_customers", 0)
    t_users  = report.get("total_bot_users", 0)

    # Context recall — what did we last discuss?
    last_topic = _last_topic(user_id)
    context_line = (
        f"\n_Last discussed: \"{last_topic}\"_"
        if last_topic else ""
    )

    # Pending alerts
    pending = []
    if l_today > 0:
        pending.append(f"{l_today} new lead(s) captured today")
    if q_today > 0:
        pending.append(f"{q_today} customer quer{'y' if q_today == 1 else 'ies'} today")
    pending_line = (
        "\n\n⚡ *Needs attention:* " + " · ".join(pending)
        if pending else ""
    )

    text = (
        f"*{greeting}, {first_name}!*\n"
        f"_{biz_name}_ · AI Persona: {persona}"
        f"{context_line}"
        f"{pending_line}\n\n"
        f"📊 *Today*\n"
        f"  Queries: *{q_today}* from *{u_today}* users\n"
        f"  New leads: *{l_today}* · Total: *{t_leads}*\n"
        f"  Verified customers: *{verified}* / {t_users} total\n\n"
        f"*Commands*\n"
        f"  /report — Full activity report\n"
        f"  /leads — Recent leads\n"
        f"  /queries — Customer queries\n"
        f"  /status — Services status\n"
        f"  /menu — Quick panel\n"
        f"  /help — BizBrain intake (for your customers)\n\n"
        f"Ask me anything about your business or customer activity."
    )
    await update.message.reply_text(text, parse_mode="Markdown")


async def _owner_show_leads(update_or_query) -> None:
    leads = get_recent_leads(10)
    if not leads:
        text = "No leads yet."
    else:
        lines = []
        for i, l in enumerate(leads, 1):
            name    = l.get("name") or l.get("telegram_id") or "Unknown"
            email   = l.get("email") or l.get("contact_info") or ""
            status  = l.get("status") or "new"
            lines.append(f"{i}. *{name}* — {_mask_email(email) if email else ''} [{status}]")
        text = "*Recent Leads*\n\n" + "\n".join(lines)
    # Can come from command or callback
    msg = getattr(update_or_query, "message", None) or getattr(update_or_query, "message", None)
    if hasattr(update_or_query, "message") and update_or_query.message:
        await update_or_query.message.reply_text(text, parse_mode="Markdown")
    else:
        await update_or_query.edit_message_text(text, parse_mode="Markdown")


async def _owner_show_queries(update_or_query) -> None:
    rows = get_recent_queries(10)
    if not rows:
        text = "No customer queries yet."
    else:
        lines = []
        for i, q in enumerate(rows, 1):
            who   = q.get("username") or q.get("telegram_id") or "?"
            query = (q.get("query") or "")[:60]
            lines.append(f"{i}. [{who}] {query}")
        text = "*Recent Customer Queries*\n\n" + "\n".join(lines)
    if hasattr(update_or_query, "message") and update_or_query.message:
        await update_or_query.message.reply_text(text, parse_mode="Markdown")
    else:
        await update_or_query.edit_message_text(text, parse_mode="Markdown")


async def _owner_show_status(msg_or_query) -> None:
    # Check Ollama
    try:
        r = requests.get(OLLAMA_URL.replace("/api/generate", ""), timeout=3)
        ollama = "running" if r.status_code < 500 else "error"
    except Exception:
        ollama = "offline"
    # Check Qdrant
    try:
        r = requests.get(f"{QDRANT_URL}/collections/{QDRANT_COLLECTION}", timeout=3)
        pts = r.json().get("result", {}).get("points_count", "?")
        qdrant = f"running ({pts} KB entries)"
    except Exception:
        qdrant = "offline"
    text = (
        f"*Bot Status*\n\n"
        f"Ollama: `{ollama}`  (model: {OLLAMA_MODEL})\n"
        f"Qdrant: `{qdrant}`\n"
        f"Dashboard: `{DASHBOARD_URL}`\n"
        f"Owner ID: `{OWNER_ID}`"
    )
    if hasattr(msg_or_query, "message") and msg_or_query.message:
        await msg_or_query.message.reply_text(text, parse_mode="Markdown")
    else:
        await msg_or_query.edit_message_text(text, parse_mode="Markdown")


async def _owner_show_report(update: Update) -> None:
    """Full detailed activity report for the owner."""
    report = get_report()
    if "error" in report:
        await update.message.reply_text(f"Could not load report: {report['error']}")
        return

    q_today  = report.get("queries_today", 0)
    u_today  = report.get("unique_users_today", 0)
    l_today  = report.get("new_leads_today", 0)
    t_leads  = report.get("total_leads", 0)
    verified = report.get("verified_customers", 0)
    t_users  = report.get("total_bot_users", 0)
    last_q   = report.get("last_queries", [])

    # Recent queries section
    if last_q:
        q_lines = "\n".join(
            f"  [{q.get('username','?')}] {q.get('query','')}"
            for q in last_q
        )
    else:
        q_lines = "  No queries yet today."

    # Recent leads
    leads = get_recent_leads(5)
    if leads:
        l_lines = "\n".join(
            f"  {i+1}. {l.get('name','?')} — {_mask_email(l.get('email') or l.get('contact_info',''))}"
            for i, l in enumerate(leads)
        )
    else:
        l_lines = "  No leads yet."

    text = (
        f"📊 *Business Activity Report*\n"
        f"_{report.get('today', 'today')}_\n\n"
        f"*Customer Queries Today*\n"
        f"  Total: *{q_today}* from *{u_today}* unique users\n\n"
        f"*Recent Questions*\n{q_lines}\n\n"
        f"*Leads*\n"
        f"  Today: *{l_today}*  ·  All-time: *{t_leads}*\n\n"
        f"*Recent Leads*\n{l_lines}\n\n"
        f"*Bot Users*\n"
        f"  Total: *{t_users}*  ·  Verified: *{verified}*\n\n"
        f"Use /leads or /queries for more detail."
    )
    await update.message.reply_text(text, parse_mode="Markdown")


# ---------------------------------------------------------------------------
# Customer onboarding handlers
# ---------------------------------------------------------------------------

async def _customer_start(update: Update, user_id: int, username: str) -> None:
    """Entry point: fast, history-based greeting + menu. No LLM wait."""
    user_rec = get_bot_user(user_id)
    step     = user_rec.get("step", STEP_NEW)

    if step == STEP_VERIFIED:
        # --- FAST PATH: known user → instant greeting + menu ---
        name        = (user_rec.get("name") or "there").split()[0]
        person_type = user_rec.get("person_type") or PERSON_VISITOR
        profile     = get_profile()
        persona     = (profile.get("ai_persona_name") or "BizNode Assistant").strip()
        biz_name    = (profile.get("business_name") or "").strip()
        greeting    = _time_greeting()
        role_greet  = _person_type_greeting(person_type, name)
        last_topic  = _last_topic(user_id)

        # Build personalized message
        lines = [f"*{greeting}!* {role_greet}"]
        if biz_name:
            lines.append(f"_{biz_name}_ — powered by *{persona}*")
        if last_topic:
            lines.append(f"\n_Last we discussed: \"{last_topic}\"_")

        # Service highlights based on role
        if person_type == PERSON_CUSTOMER:
            lines.append(
                "\nI'm here 24/7 to help with your needs — pricing, "
                "support, or any business question."
            )
        elif person_type == PERSON_INVESTOR:
            lines.append("\nI can show you real-time business metrics and growth data.")
        elif person_type in (PERSON_VISITOR, PERSON_GUEST):
            lines.append(
                "\nExplore our services below, or ask me anything! "
                "Free to use — get DZIT credits at https://dzit.1bz.biz "
                "for advanced AI bot features."
            )

        lines.append("\n*Select from the menu below:*")

        await update.message.reply_text(
            "\n".join(lines),
            parse_mode="Markdown",
            reply_markup=_main_menu(person_type),
            disable_web_page_preview=True,
        )
        return

    if step in (STEP_AWAITING_NAME, STEP_NEW, ""):
        await _onboard_start(update, user_id, username)
    elif step == STEP_AWAITING_BUSINESS:
        await update.message.reply_text("What business or company are you from?")
    elif step == STEP_AWAITING_EMAIL:
        await update.message.reply_text(
            "Please share your *email address* for verification.",
            parse_mode="Markdown",
        )
    elif step == STEP_AWAITING_OTP:
        email = user_rec.get("email", "your email")
        await update.message.reply_text(
            f"I sent a verification code to *{email}*. Please enter it here.",
            parse_mode="Markdown",
        )


async def _onboard_start(update: Update, user_id: int, username: str) -> None:
    profile  = get_profile()
    persona  = (profile.get("ai_persona_name") or "BizNode Assistant").strip()
    biz_name = (profile.get("business_name") or "").strip()
    greeting = (profile.get("greeting_message") or "").strip()

    if greeting:
        intro = greeting
    elif biz_name:
        intro = f"Hi! I'm *{persona}*, the AI assistant for *{biz_name}*."
    else:
        intro = f"Hi! I'm *{persona}*, your AI business assistant."

    intro += (
        "\n\nBefore we begin, I'd like to know you a little.\n\n"
        "*What's your name?*"
    )
    update_bot_user(user_id, step=STEP_AWAITING_NAME, username=username)
    await update.message.reply_text(intro, parse_mode="Markdown")


async def _handle_awaiting_name(update: Update, user_id: int, text: str) -> None:
    name = text.strip()[:100]
    update_bot_user(user_id, name=name, step=STEP_AWAITING_BUSINESS)
    await update.message.reply_text(
        f"Nice to meet you, *{name}*!\n\n"
        "What business or company are you from? _(type *personal* if none)_",
        parse_mode="Markdown",
    )


async def _handle_awaiting_business(update: Update, user_id: int, text: str) -> None:
    business = text.strip()[:200]
    update_bot_user(user_id, business=business, step=STEP_AWAITING_EMAIL)
    await update.message.reply_text(
        f"Got it — *{business}*.\n\n"
        "Please share your *email address* so I can verify your identity.\n"
        "_Type *skip* if you'd rather do this later._",
        parse_mode="Markdown",
    )


async def _handle_awaiting_email(update: Update, user_id: int,
                                   text: str, user_rec: dict) -> None:
    raw = text.strip()

    # Allow user to skip email verification
    if raw.lower() in ("skip", "none", "no", "later", "pass", "na", "n/a"):
        logger.info("User %s skipped email verification", user_id)
        create_lead(
            name        = user_rec.get("name", ""),
            business    = user_rec.get("business", ""),
            email       = "",
            telegram_id = user_id,
        )
        update_bot_user(user_id, email="", step=STEP_VERIFIED, verified=1,
                        person_type=PERSON_CUSTOMER)
        first_name = (user_rec.get("name") or "there").split()[0]
        await update.message.reply_text(
            f"No problem, *{first_name}*! You can add your email later.\n\n"
            "What best describes you?",
            parse_mode="Markdown",
            reply_markup=InlineKeyboardMarkup([
                [InlineKeyboardButton("I'm a Customer", callback_data="role:customer")],
                [InlineKeyboardButton("I'm a Partner / Business", callback_data="role:partner")],
            ]),
        )
        return

    # Extract email even if surrounded by extra text
    email = _extract_email(raw)
    if not email:
        await update.message.reply_text(
            "I couldn't find a valid email in your message. Please send just the email address.\n"
            "Example: _name@company.com_\n\n"
            "_Type *skip* if you'd rather do this later._",
            parse_mode="Markdown",
        )
        return
    name = user_rec.get("name") or "there"
    otp_result = send_otp(user_id, email, name)
    if otp_result == "ok":
        update_bot_user(user_id, email=email, step=STEP_AWAITING_OTP)
        await update.message.reply_text(
            f"I've sent a 6-digit verification code to *{email}*.\n\n"
            "Please check your inbox and enter the code here.\n"
            "_Code is valid for 15 minutes._",
            parse_mode="Markdown",
        )
    elif otp_result == "smtp_not_configured":
        # SMTP not set up — skip OTP and auto-verify so customer isn't stuck
        logger.warning("SMTP not configured — auto-verifying user %s (%s)", user_id, email)
        create_lead(
            name        = user_rec.get("name", ""),
            business    = user_rec.get("business", ""),
            email       = email,
            telegram_id = user_id,
        )
        update_bot_user(user_id, email=email, step=STEP_VERIFIED, verified=1,
                        person_type=PERSON_CUSTOMER)
        first_name = (user_rec.get("name") or "there").split()[0]
        await update.message.reply_text(
            f"Welcome, *{first_name}*! You're all set. \n\n"
            "Email verification was skipped because the email service "
            "hasn't been configured yet.\n\n"
            "What best describes you?",
            parse_mode="Markdown",
            reply_markup=InlineKeyboardMarkup([
                [InlineKeyboardButton("I'm a Customer", callback_data="role:customer")],
                [InlineKeyboardButton("I'm a Partner / Business", callback_data="role:partner")],
            ]),
        )
    else:
        await update.message.reply_text(
            "Couldn't send the verification email right now. "
            "Please try again in a moment, or contact the business owner."
        )


async def _handle_awaiting_otp(update: Update, user_id: int,
                                 text: str, user_rec: dict) -> None:
    code = text.strip().replace(" ", "")
    verified, msg = verify_otp(user_id, code)
    if verified:
        create_lead(
            name        = user_rec.get("name", ""),
            business    = user_rec.get("business", ""),
            email       = user_rec.get("email", ""),
            telegram_id = user_id,
        )
        update_bot_user(user_id, step=STEP_VERIFIED, verified=1,
                        person_type=PERSON_CUSTOMER)
        name = (user_rec.get("name") or "there").split()[0]
        await update.message.reply_text(
            f"You're verified, *{name}*! Welcome. ✅\n\n"
            "One last thing — what best describes you?",
            parse_mode="Markdown",
            reply_markup=_role_selection_menu(),
        )
    else:
        await update.message.reply_text(
            f"That code didn't match. {msg}\n\n"
            "Try again, or /start to restart."
        )


# ---------------------------------------------------------------------------
# Telegram command handlers
# ---------------------------------------------------------------------------

async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    user_id  = update.effective_user.id
    username = update.effective_user.username or ""

    if is_owner(user_id):
        logger.info("OWNER /start — telegram_id=%s", user_id)
        await _owner_start(update)
        return

    if is_blocked(user_id):
        await update.message.reply_text(
            "Your access has been restricted. Contact the business owner for help."
        )
        return

    logger.info("CUSTOMER /start — telegram_id=%s username=%s", user_id, username)
    await _customer_start(update, user_id, username)


async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Show command guide — different for owner vs customer."""
    if update.effective_chat and update.effective_chat.type != "private":
        return

    user_id = update.effective_user.id

    if is_owner(user_id):
        await update.message.reply_text(
            "*BizNode — Owner Command Guide*\n\n"
            "*Business Management:*\n"
            "/report — Daily activity report\n"
            "/leads — View recent leads\n"
            "/queries — Recent customer queries\n"
            "/status — Bot & service status\n"
            "/menu — Owner dashboard menu\n\n"
            "*Knowledge Base:*\n"
            "/kbadd `Title | Content` — Add KB entry\n"
            "/kbremove `ID or search` — Remove KB entry\n"
            "/memory — KB entry count\n\n"
            "*Group & Marketing:*\n"
            "/postgroup `message` — Post to group\n"
            "/affiliate — Your affiliate links\n\n"
            "*Email & Communication:*\n"
            "Just say: _\"send email to user@email.com about topic\"_\n"
            "Or: _\"mail to user@email.com about follow up\"_\n\n"
            "*Payments & Licenses:*\n"
            "/services — Manage services\n"
            "/purchase\\_status `REF` — Check purchase\n\n"
            "*AI Assistant:*\n"
            "Ask anything naturally — I can look up leads, "
            "compose emails, check status, search KB, and more.\n\n"
            "*Update Packets:*\n"
            "/sendupdate `type ver tier summary` — Create & send update PDF\n"
            "/setpayload `{json}` — Set payload for next update\n"
            "/applyupdate `OTP` — Confirm blockchain update\n\n"
            "*Ad Marketplace:*\n"
            "/bo — Marketplace help\n"
            "/bo submit `title` — Submit ad proposal\n"
            "/bo confirm `id otp` — OTP confirmation\n"
            "/bo quote `id amount` — Quote for proposal\n"
            "/bo accept `id` — Accept consolidated quote\n"
            "/bo status `id` — Proposal/campaign status\n"
            "/bo campaigns — Active campaigns\n"
            "/bo stats — Marketplace statistics\n\n"
            "*Other:*\n"
            "/version — Check BizNode version & updates\n"
            "/bizbrain — Generate AI business profile\n"
            "/clear — Clear conversation memory\n"
            "/role — View your role\n\n"
            "_Tip: You can talk to me like a real assistant. "
            "I understand natural language, not just commands._",
            parse_mode="Markdown",
        )
    else:
        await update.message.reply_text(
            "*BizNode — Help Guide*\n\n"
            "*Getting Started:*\n"
            "/start — Begin or restart\n"
            "/menu — Browse services & features\n"
            "/role — Change your role\n\n"
            "*Services & Payments:*\n"
            "/services — View available services\n"
            "/buy\\_license — Purchase a BizNode license\n"
            "/purchase\\_status `REF` — Check purchase status\n"
            "/wallet `0xAddress` — Register crypto wallet\n"
            "/verify\\_wallet — Verify crypto payment\n"
            "/affiliate — Join affiliate program\n\n"
            "*Transactions:*\n"
            "/transact — Transaction options\n"
            "/pay — Make a payment\n"
            "/negotiate — Negotiate a deal\n\n"
            "*Ad Marketplace:*\n"
            "/bo — Browse business opportunities\n"
            "/bo submit `title` — Submit ad proposal\n"
            "/bo quote `id amount` — Quote for a proposal\n"
            "/bo status `id` — Check status\n\n"
            "*Conversation:*\n"
            "/clear — Clear chat memory\n"
            "/memory — Check knowledge base\n"
            "/bizbrain — Generate AI business profile\n\n"
            "*About:*\n"
            "/ceo — Founder profile\n"
            "/version — Check BizNode version\n\n"
            "_Ask me anything about the business — "
            "I'm here to help!_",
            parse_mode="Markdown",
        )


async def cmd_bizbrain(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """4-step BizBrain business intake (only in private chats)."""
    if update.effective_chat and update.effective_chat.type != "private":
        return

    user_id = update.effective_user.id
    _help_intake[user_id] = {"step": 1}
    await update.message.reply_text(
        "*BizBrain Business Profile*\n\n"
        "I'll ask you 4 quick questions to generate your personalised AI business profile.\n\n"
        f"*Question 1 of 4:* {HELP_STEP_QUESTIONS[1]}",
        parse_mode="Markdown",
    )


async def cmd_menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    user_id = update.effective_user.id
    if is_owner(user_id):
        await update.message.reply_text(
            "*BizNode Owner Menu*\n\n"
            "Tap any category below:",
            parse_mode="Markdown",
            reply_markup=_owner_menu(),
        )
        return
    user_rec = get_bot_user(user_id)
    if user_rec.get("step") != STEP_VERIFIED:
        await update.message.reply_text(
            "Please complete verification first. Type /start to begin."
        )
        return
    await update.message.reply_text("What can I help you with?",
                                     reply_markup=_main_menu())


async def cmd_report(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Owner only: detailed activity report."""
    if not is_owner(update.effective_user.id):
        await update.message.reply_text("This command is for the bot owner only.")
        return
    await _owner_show_report(update)


async def cmd_leads(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Owner only: show recent leads."""
    if not is_owner(update.effective_user.id):
        await update.message.reply_text("This command is for the bot owner only.")
        return
    await _owner_show_leads(update)


async def cmd_queries(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Owner only: show recent customer queries."""
    if not is_owner(update.effective_user.id):
        await update.message.reply_text("This command is for the bot owner only.")
        return
    await _owner_show_queries(update)


async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Owner only: show bot/service status."""
    if not is_owner(update.effective_user.id):
        await update.message.reply_text("This command is for the bot owner only.")
        return
    await _owner_show_status(update)


async def cmd_ceo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Public founder profile with logo."""
    import pathlib
    logo_path = pathlib.Path(__file__).resolve().parent.parent.parent / "logos" / "biznodelogo.png"
    caption = (
        "*Founder & CEO*\n"
        "1BZ DZIT DAO LLC\n\n"
        "Building BizNode\u2122 \u2014 autonomous AI business infrastructure "
        "that puts the power of automation in every business owner's hands.\n\n"
        "*Connect:*\n"
        "[LinkedIn](https://www.linkedin.com/in/1bz/)\n\n"
        "*Company:*\n"
        "1BZ DZIT DAO LLC\n"
        "biznode@1bz.biz\n"
        "[biznode.1bz.biz](https://biznode.1bz.biz)\n"
        "[GitHub](https://github.com/1bzdzit/biznode)"
    )
    try:
        if logo_path.exists():
            with open(str(logo_path), "rb") as photo:
                await update.message.reply_photo(
                    photo=photo,
                    caption=caption,
                    parse_mode="Markdown",
                )
            return
    except Exception as exc:
        logger.warning("Failed to send logo: %s", exc)
    # Fallback to text only
    await update.message.reply_text(caption, parse_mode="Markdown", disable_web_page_preview=True)


async def cmd_version(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Show current BizNode version and check for updates."""
    import pathlib
    local_ver = "unknown"
    try:
        vf = pathlib.Path(__file__).resolve().parent.parent.parent / "VERSION"
        if vf.exists():
            local_ver = vf.read_text().strip()
    except Exception:
        pass

    server_ver = "?"
    update_available = False
    try:
        r = requests.get("https://biznode.1bz.biz/api/version.php", timeout=10)
        if r.status_code == 200:
            data = r.json()
            server_ver = data.get("version", "?")
            if server_ver != local_ver and server_ver != "?":
                update_available = True
    except Exception:
        pass

    msg = (
        f"*BizNode Version*\n\n"
        f"Running : `{local_ver}`\n"
        f"Latest  : `{server_ver}`\n"
    )
    if update_available:
        msg += (
            f"\n*Update available!*\n"
            f"Run: `python scripts/update_biznode.py`\n"
            f"Or: `python scripts/update_biznode.py --check` to preview"
        )
    else:
        msg += "\nYou're up to date."

    await update.message.reply_text(msg, parse_mode="Markdown")


async def cmd_role(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Let user select their role/person type."""
    user_id = update.effective_user.id
    if is_owner(user_id):
        await update.message.reply_text("You're the owner — role is fixed.")
        return
    user_rec = get_bot_user(user_id)
    current = user_rec.get("person_type") or PERSON_VISITOR
    label = PERSON_TYPE_LABELS.get(current, current)
    await update.message.reply_text(
        f"Your current role: *{label}*\n\nSelect your role:",
        parse_mode="Markdown",
        reply_markup=_role_selection_menu(),
    )


async def cmd_clear(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    clear_history(update.effective_user.id)
    await update.message.reply_text("Conversation memory cleared.")


async def cmd_memory(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    try:
        r = requests.get(f"{QDRANT_URL}/collections/{QDRANT_COLLECTION}", timeout=5)
        count = r.json().get("result", {}).get("points_count", "unknown")
        await update.message.reply_text(f"Knowledge base: {count} entries stored.")
    except Exception as exc:
        await update.message.reply_text(f"Memory unavailable: {exc}")


# ---------------------------------------------------------------------------
# Transaction commands
# ---------------------------------------------------------------------------

async def cmd_transact(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Show transaction options based on person type. /transact"""
    user_id = update.effective_user.id
    user_rec = get_bot_user(user_id)
    person_type = user_rec.get("person_type") or PERSON_VISITOR

    data = _dash_get(f"/api/transactions/suggested/{person_type}")
    if not isinstance(data, list) or not data:
        data = [
            {"type": "service_inquiry", "label": "Service Inquiry"},
            {"type": "consultation", "label": "Book Consultation"},
        ]

    keyboard = []
    for item in data:
        keyboard.append([InlineKeyboardButton(
            item["label"], callback_data=f"txn:{item['type']}"
        )])
    keyboard.append([InlineKeyboardButton("My Transactions", callback_data="txn:history")])

    name = (user_rec.get("name") or "there").split()[0]
    await update.message.reply_text(
        f"*{name}, what would you like to do?*\n\n"
        "Select a transaction type:",
        parse_mode="Markdown",
        reply_markup=InlineKeyboardMarkup(keyboard),
    )


async def cmd_pay(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Show payment options. /pay"""
    user_id = update.effective_user.id
    user_rec = get_bot_user(user_id)

    keyboard = [
        [InlineKeyboardButton("Pay with Card (Stripe)", callback_data="pay:stripe")],
        [InlineKeyboardButton("Pay with Crypto (MATIC)", callback_data="pay:crypto")],
        [InlineKeyboardButton("View Payment History", callback_data="pay:history")],
    ]

    await update.message.reply_text(
        "*Payment Options*\n\n"
        "Choose your preferred payment method:",
        parse_mode="Markdown",
        reply_markup=InlineKeyboardMarkup(keyboard),
    )


async def cmd_wallet(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Register or check crypto wallet. /wallet <address>"""
    user_id = update.effective_user.id
    args = context.args

    if not args:
        # Show wallet status
        info = _dash_get(f"/api/crypto/wallet/{user_id}")
        if info.get("registered"):
            wallet = info.get("wallet", "")
            verified = "Yes" if info.get("verified") else "No"
            await update.message.reply_text(
                f"*Your Crypto Wallet*\n\n"
                f"Address: `{wallet}`\n"
                f"Verified: *{verified}*\n\n"
                f"Use /verify\\_wallet after sending payment.",
                parse_mode="Markdown",
            )
        else:
            await update.message.reply_text(
                "*Register Your Crypto Wallet*\n\n"
                "Usage: `/wallet 0xYourPolygonAddress`\n\n"
                "_Polygon (MATIC) network only._",
                parse_mode="Markdown",
            )
        return

    wallet_address = args[0].strip()
    result = _dash_post("/api/crypto/register-wallet", {
        "telegram_id": str(user_id),
        "wallet_address": wallet_address,
    })

    if result.get("ok"):
        biznode_wallet = result.get("biznode_wallet", "")
        amount = result.get("verification_amount", 0.01)
        if biznode_wallet:
            await update.message.reply_text(
                f"Wallet registered: `{wallet_address}`\n\n"
                f"To verify, send *{amount} MATIC* to:\n"
                f"`{biznode_wallet}`\n\n"
                f"Then type /verify\\_wallet",
                parse_mode="Markdown",
            )
        else:
            await update.message.reply_text(
                f"Wallet registered: `{wallet_address}`\n\n"
                "Crypto payment details will be provided by the owner.",
                parse_mode="Markdown",
            )
    else:
        await update.message.reply_text(result.get("error", "Registration failed."))


async def cmd_verify_wallet(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Verify crypto wallet payment. /verify_wallet"""
    user_id = update.effective_user.id
    result = _dash_post("/api/crypto/verify-wallet", {"telegram_id": str(user_id)})

    if result.get("ok"):
        await update.message.reply_text(
            "Your wallet is now verified! All future payments from this wallet "
            "will be tracked automatically.",
        )
    elif result.get("mode") == "manual":
        await update.message.reply_text(
            f"{result.get('message', 'Verification pending.')}\n\n"
            "The owner will confirm your payment shortly.",
        )
    else:
        await update.message.reply_text(result.get("error", "Verification failed."))


async def cmd_affiliate(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Show affiliate/partner links. /affiliate"""
    data = _dash_get("/api/affiliate-links", active_only=True)
    if not isinstance(data, list) or not data:
        await update.message.reply_text(
            "No affiliate links available right now.\n\n"
            "Check back later for partner deals and recommendations!"
        )
        return

    lines = ["*Recommended Partners & Resources:*\n"]
    keyboard = []
    for link in data[:8]:
        title = link.get("title", "Link")
        url = link.get("url", "")
        desc = link.get("description", "")
        commission = link.get("commission", "")
        line = f"- *{title}*"
        if desc:
            line += f" — {desc}"
        if commission:
            line += f" _{commission}_"
        lines.append(line)
        keyboard.append([InlineKeyboardButton(title, url=url)])

    await update.message.reply_text(
        "\n".join(lines),
        parse_mode="Markdown",
        reply_markup=InlineKeyboardMarkup(keyboard) if keyboard else None,
        disable_web_page_preview=True,
    )


async def cmd_services(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Show available services with prices. /services"""
    data = _dash_get("/api/services")
    if not isinstance(data, list) or not data:
        await update.message.reply_text("No services listed yet. Contact us for details!")
        return

    lines = ["*Our Services:*\n"]
    keyboard = []
    for svc in data:
        if not svc.get("is_active"):
            continue
        name = svc.get("name", "Service")
        price = svc.get("price", 0)
        currency = svc.get("currency", "USD")
        desc = svc.get("description", "")
        duration = svc.get("duration", "")
        line = f"*{name}* — {currency} {price:.2f}"
        if duration:
            line += f" / {duration}"
        if desc:
            line += f"\n  _{desc}_"
        lines.append(line)
        keyboard.append([InlineKeyboardButton(
            f"Buy: {name} ({currency} {price:.2f})",
            callback_data=f"buy:{svc['id']}"
        )])

    await update.message.reply_text(
        "\n\n".join(lines),
        parse_mode="Markdown",
        reply_markup=InlineKeyboardMarkup(keyboard) if keyboard else None,
    )


async def cmd_negotiate(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Start price negotiation. /negotiate <service> <your_price>"""
    user_id = update.effective_user.id
    args = context.args

    if not args or len(args) < 2:
        await update.message.reply_text(
            "*Price Negotiation*\n\n"
            "Usage: `/negotiate ServiceName YourPrice`\n"
            "Example: `/negotiate WebDev 500`\n\n"
            "I'll present your offer to the business owner and get back to you!",
            parse_mode="Markdown",
        )
        return

    service_name = " ".join(args[:-1])
    try:
        desired_price = float(args[-1])
    except ValueError:
        await update.message.reply_text("Please provide a valid price number.")
        return

    user_rec = get_bot_user(user_id)
    person_type = user_rec.get("person_type") or PERSON_CUSTOMER
    name = user_rec.get("name") or "Customer"
    email = user_rec.get("email") or ""

    result = _dash_post("/api/transactions", {
        "telegram_id": str(user_id),
        "person_type": person_type,
        "transaction_type": "price_negotiation",
        "title": f"Price negotiation: {service_name}",
        "description": f"{name} requests ${desired_price:.2f} for {service_name}",
        "amount": desired_price,
        "currency": "USD",
        "metadata": {"service_name": service_name, "desired_price": desired_price,
                      "customer_name": name, "customer_email": email},
    })

    if result.get("reference_no"):
        ref = result["reference_no"]
        await update.message.reply_text(
            f"*Negotiation Started!* (Ref: `{ref}`)\n\n"
            f"Service: *{service_name}*\n"
            f"Your offer: *${desired_price:.2f}*\n\n"
            f"I've notified the business owner. You'll hear back soon!",
            parse_mode="Markdown",
        )
        # Notify owner
        try:
            import os as _os
            from services.email_service import EmailService
            es = EmailService()
            owner_email = _os.getenv("OWNER_EMAIL", "")
            if owner_email:
                es.send_email(
                    owner_email,
                    f"Price Negotiation Request: {service_name} — {ref}",
                    f"Customer {name} ({email}) wants {service_name} at ${desired_price:.2f}.\n"
                    f"Reference: {ref}\n\nReply via dashboard or email.",
                )
        except Exception:
            pass
    else:
        await update.message.reply_text("Could not start negotiation. Please try again.")


# ---------------------------------------------------------------------------
# License purchase commands
# ---------------------------------------------------------------------------

# Pending crypto OTP sessions: telegram_id -> {reference_no, step}
_PENDING_CRYPTO_PURCHASE: dict = {}


async def cmd_buy_license(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Purchase a BizNode license. /buy_license [basic|pro]"""
    user_id = update.effective_user.id
    args = context.args

    license_type = "basic"
    if args and args[0].lower() in ("pro", "basic"):
        license_type = args[0].lower()

    keyboard = [
        [InlineKeyboardButton(
            "Buy Basic License (Card)",
            callback_data=f"license:stripe:basic",
        )],
        [InlineKeyboardButton(
            "Buy Basic License (Crypto)",
            callback_data=f"license:crypto:basic",
        )],
        [InlineKeyboardButton(
            "Buy Pro License (Card)",
            callback_data=f"license:stripe:pro",
        )],
        [InlineKeyboardButton(
            "Buy Pro License (Crypto)",
            callback_data=f"license:crypto:pro",
        )],
    ]

    await update.message.reply_text(
        "*BizNode License Purchase*\n\n"
        "*Basic License* — $200 USD or 200 BZeUSD\n"
        "  1 bot, core features, software key\n\n"
        "*Pro License* — $499 USD or 499 BZeUSD\n"
        "  Up to 5 bots, all features, USB hardware key\n\n"
        "Choose your license and payment method:",
        parse_mode="Markdown",
        reply_markup=InlineKeyboardMarkup(keyboard),
    )


async def _handle_license_callback(query, user_id: int, sub: str) -> None:
    """Handle license: callbacks for license purchase."""
    parts = sub.split(":")
    if len(parts) != 2:
        await query.message.reply_text("Invalid selection.")
        return

    payment_method, license_type = parts

    # Get user info
    user_rec = get_bot_user(user_id)
    email = (user_rec or {}).get("email", "")
    name = (user_rec or {}).get("name", "")

    if not email:
        await query.message.reply_text(
            "You need to verify your email first before purchasing.\n\n"
            "Please complete the registration process (/start) and verify your email.",
        )
        return

    if payment_method == "stripe":
        # Create Stripe license purchase
        result = _dash_post("/api/license-purchase/stripe", {
            "email": email,
            "customer_name": name,
            "telegram_id": str(user_id),
            "license_type": license_type,
        })
        if result.get("ok"):
            payment_url = result.get("payment_url", "")
            ref = result.get("reference_no", "")
            tier_label = "Basic" if license_type == "basic" else "Pro"
            amount = result.get("amount", 0)
            keyboard = [[InlineKeyboardButton(
                f"Pay ${amount:.0f} Now",
                url=payment_url,
            )]]
            await query.message.reply_text(
                f"*BizNode {tier_label} License*\n\n"
                f"Amount: *${amount:.2f} USD*\n"
                f"Reference: `{ref}`\n\n"
                f"Click below to complete your payment:",
                parse_mode="Markdown",
                reply_markup=InlineKeyboardMarkup(keyboard),
            )
        else:
            await query.message.reply_text(
                f"Could not create payment: {result.get('error', 'Unknown error')}"
            )

    elif payment_method == "crypto":
        # Initiate crypto purchase — sends OTP
        result = _dash_post("/api/license-purchase/crypto/initiate", {
            "email": email,
            "customer_name": name,
            "telegram_id": str(user_id),
            "license_type": license_type,
        })
        if result.get("ok"):
            ref = result.get("reference_no", "")
            wallet = result.get("wallet_address", "")
            token = result.get("token", "BZeUSD")
            amount = result.get("amount", 0)
            otp_expires = result.get("otp_expires_minutes", 15)

            # Store pending state
            _PENDING_CRYPTO_PURCHASE[user_id] = {
                "reference_no": ref,
                "step": "awaiting_otp",
            }

            tier_label = "Basic" if license_type == "basic" else "Pro"
            await query.message.reply_text(
                f"*BizNode {tier_label} License — Crypto Payment*\n\n"
                f"*Step 1:* An OTP has been sent to your email.\n"
                f"*Step 2:* Send *${amount:.2f}* in {token} to:\n"
                f"`{wallet}`\n"
                f"Network: *Polygon*\n\n"
                f"*Step 3:* After payment, reply with:\n"
                f"`/verify_purchase {ref} YOUR_OTP YOUR_TX_HASH`\n\n"
                f"OTP expires in *{otp_expires} minutes*.\n"
                f"Reference: `{ref}`",
                parse_mode="Markdown",
            )
        else:
            await query.message.reply_text(
                f"Could not initiate payment: {result.get('error', 'Unknown error')}"
            )


async def cmd_verify_purchase(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Verify a crypto license purchase. /verify_purchase <ref> <otp> <tx_hash>"""
    user_id = update.effective_user.id
    args = context.args

    if not args or len(args) < 3:
        await update.message.reply_text(
            "*Verify Crypto Purchase*\n\n"
            "Usage: `/verify_purchase REFERENCE OTP TX_HASH`\n\n"
            "Example:\n"
            "`/verify_purchase BN-20260310-0001 123456 0xabc...def`\n\n"
            "You received the reference and OTP in your email.",
            parse_mode="Markdown",
        )
        return

    reference_no = args[0].strip()
    otp_code = args[1].strip()
    tx_hash = args[2].strip()

    await update.message.reply_text("Verifying your payment... Please wait.")

    result = _dash_post("/api/license-purchase/crypto/verify", {
        "reference_no": reference_no,
        "otp_code": otp_code,
        "transaction_hash": tx_hash,
    })

    if result.get("ok"):
        license_type = result.get("license_type", "basic")
        if license_type == "basic":
            await update.message.reply_text(
                "*Payment Verified!*\n\n"
                "Your BizNode Basic license key has been sent to your email.\n\n"
                "Follow the instructions in the email to activate your license.",
                parse_mode="Markdown",
            )
        else:
            await update.message.reply_text(
                "*Payment Verified!*\n\n"
                "Your BizNode Pro license is being prepared.\n"
                "The owner will ship your USB license device. "
                "You'll receive an email with tracking information.",
                parse_mode="Markdown",
            )
        # Clean up pending state
        _PENDING_CRYPTO_PURCHASE.pop(user_id, None)
    else:
        error = result.get("error", "Verification failed")
        await update.message.reply_text(
            f"*Verification Failed*\n\n{error}",
            parse_mode="Markdown",
        )


async def cmd_purchase_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Check license purchase status. /purchase_status <reference>"""
    args = context.args
    if not args:
        await update.message.reply_text(
            "Usage: `/purchase_status BN-XXXXXXXX-XXXX`",
            parse_mode="Markdown",
        )
        return

    ref = args[0].strip()
    result = _dash_get(f"/api/license-purchase/{ref}")

    if result.get("ok"):
        status = result.get("status", "unknown")
        ltype = result.get("license_type", "basic")
        method = result.get("payment_method", "")
        created = result.get("created_at", "")
        tier_label = "Basic" if ltype == "basic" else "Pro"
        await update.message.reply_text(
            f"*Purchase Status*\n\n"
            f"Reference: `{ref}`\n"
            f"License: *{tier_label}*\n"
            f"Payment: {method.upper()}\n"
            f"Status: *{status.upper()}*\n"
            f"Created: {created}",
            parse_mode="Markdown",
        )
    else:
        await update.message.reply_text(result.get("error", "Purchase not found."))


# ---------------------------------------------------------------------------
# /kbadd, /kbremove — Owner knowledge base management
# ---------------------------------------------------------------------------

async def cmd_kbadd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Add an entry to the support knowledge base. Owner only.
    Usage: /kbadd <title> | <content>
    Or:    /kbadd <content>  (title auto-generated)
    """
    user_id = update.effective_user.id
    if not is_owner(user_id):
        await update.message.reply_text("This command is for the business owner only.")
        return

    raw = " ".join(context.args) if context.args else ""
    if not raw.strip():
        await update.message.reply_text(
            "Usage:\n"
            "`/kbadd My Title | The article content here`\n"
            "or just:\n"
            "`/kbadd The content to add to knowledge base`",
            parse_mode="Markdown",
        )
        return

    # Split title | content, or use first 60 chars as title
    if "|" in raw:
        title, content = raw.split("|", 1)
        title = title.strip()
        content = content.strip()
    else:
        content = raw.strip()
        title = content[:60] + ("..." if len(content) > 60 else "")

    if not content:
        await update.message.reply_text("Content cannot be empty.")
        return

    thinking = await update.message.reply_text("Adding to knowledge base...")
    try:
        vec = embed_text(f"{title} {content}")
        if not vec:
            await thinking.edit_text("Failed to generate embedding. Is Ollama running?")
            return

        import hashlib, struct
        h = hashlib.sha256(f"kb:{title}:{content}".encode()).digest()
        point_id = struct.unpack(">Q", h[:8])[0]

        requests.put(
            f"{QDRANT_URL}/collections/biznode_support/points",
            json={"points": [{
                "id": point_id,
                "vector": vec,
                "payload": {
                    "title": title,
                    "content": content,
                    "category": "owner_added",
                    "source": "telegram_kbadd",
                },
            }]},
            timeout=10,
        )
        await thinking.edit_text(
            f"Added to Knowledge Base\n\n"
            f"Title: *{title}*\n"
            f"ID: `{point_id}`\n\n"
            f"_Use /kbremove {point_id} to delete._",
            parse_mode="Markdown",
        )
        logger.info("KB added by owner: id=%s title=%s", point_id, title)
    except Exception as exc:
        logger.error("KB add failed: %s", exc)
        await thinking.edit_text(f"Failed to add: {exc}")


async def cmd_kbremove(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Remove a knowledge base entry. Owner only.
    Usage: /kbremove <id>           — remove by point ID
    Usage: /kbremove <search text>  — find & remove by content match
    """
    user_id = update.effective_user.id
    if not is_owner(user_id):
        await update.message.reply_text("This command is for the business owner only.")
        return

    raw = " ".join(context.args) if context.args else ""
    if not raw.strip():
        await update.message.reply_text(
            "Usage:\n"
            "`/kbremove 12345678`  — remove by ID\n"
            "`/kbremove backup strategy`  — search and remove",
            parse_mode="Markdown",
        )
        return

    thinking = await update.message.reply_text("Searching knowledge base...")

    # Try as numeric ID first
    try:
        point_id = int(raw.strip())
        requests.post(
            f"{QDRANT_URL}/collections/biznode_support/points/delete",
            json={"points": [point_id]},
            timeout=10,
        )
        await thinking.edit_text(f"Removed KB entry `{point_id}`.", parse_mode="Markdown")
        logger.info("KB removed by owner: id=%s", point_id)
        return
    except (ValueError, TypeError):
        pass

    # Search by text
    try:
        vec = embed_text(raw.strip())
        if not vec:
            await thinking.edit_text("Failed to generate embedding for search.")
            return

        r = requests.post(
            f"{QDRANT_URL}/collections/biznode_support/points/search",
            json={"vector": vec, "limit": 3, "with_payload": True},
            timeout=10,
        )
        r.raise_for_status()
        results = r.json().get("result", [])

        if not results:
            await thinking.edit_text("No matching KB entries found.")
            return

        # Show top match and confirm removal
        top = results[0]
        pid = top["id"]
        payload = top.get("payload", {})
        title = payload.get("title", "Untitled")
        score = top.get("score", 0)

        requests.post(
            f"{QDRANT_URL}/collections/biznode_support/points/delete",
            json={"points": [pid]},
            timeout=10,
        )
        await thinking.edit_text(
            f"Removed best match (score {score:.2f}):\n\n"
            f"*{title}*\n"
            f"ID: `{pid}`",
            parse_mode="Markdown",
        )
        logger.info("KB removed by owner (search): id=%s title=%s", pid, title)
    except Exception as exc:
        logger.error("KB remove failed: %s", exc)
        await thinking.edit_text(f"Failed to remove: {exc}")


# ---------------------------------------------------------------------------
# /postgroup — Owner posts message to Telegram group
# ---------------------------------------------------------------------------

async def cmd_postgroup(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Post a message to the business Telegram group. Owner only.
    Usage: /postgroup <message>
    """
    user_id = update.effective_user.id
    if not is_owner(user_id):
        await update.message.reply_text("This command is for the business owner only.")
        return

    if not TELEGRAM_GROUP_ID:
        await update.message.reply_text("No Telegram group configured (TELEGRAM_GROUP_ID not set).")
        return

    raw = " ".join(context.args) if context.args else ""
    if not raw.strip():
        await update.message.reply_text(
            "Usage: `/postgroup Your message to post in the group`",
            parse_mode="Markdown",
        )
        return

    try:
        await context.bot.send_message(
            chat_id=TELEGRAM_GROUP_ID,
            text=raw.strip(),
            parse_mode="Markdown",
        )
        await update.message.reply_text("Posted to group.")
        logger.info("Owner posted to group: %s", raw[:80])
    except Exception as exc:
        logger.error("Group post failed: %s", exc)
        await update.message.reply_text(f"Failed to post: {exc}")


# ---------------------------------------------------------------------------
# Update Packet System (Admin only)
# ---------------------------------------------------------------------------

async def cmd_sendupdate(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Create and send an update packet PDF to the Telegram group or individual node.

    Usage:
        /sendupdate <type> <version> <tier> <summary>
        /sendupdate updatekb 1.2.1 all Added new KB articles about email templates

    Types: updatetemplate, updatemenu, updateworkflow, updatebiz, updatekb,
           updatetables, updatedzit (pro), updatebzeusd (pro)
    Tier:  all, basic, pro

    For targeted (individual node): /sendupdate <type> <version> <tier> --node <node_id> <summary>
    For email delivery: /sendupdate <type> <version> <tier> --email <email> <summary>
    """
    user_id = update.effective_user.id
    if not is_owner(user_id):
        await update.message.reply_text("This command is for the business owner only.")
        return

    args = context.args or []
    if len(args) < 4:
        await update.message.reply_text(
            "*Update Packet Creator*\n\n"
            "Usage: `/sendupdate <type> <version> <tier> <summary>`\n\n"
            "*Types:*\n"
            "  `updatetemplate` - Email/UI templates\n"
            "  `updatemenu` - Menu & navigation\n"
            "  `updateworkflow` - Workflow engine\n"
            "  `updatebiz` - Business config\n"
            "  `updatekb` - Knowledge base\n"
            "  `updatetables` - DB migrations\n"
            "  `updatedzit` - DZIT credits (Pro)\n"
            "  `updatebzeusd` - BZeUSD balance (Pro)\n\n"
            "*Tier:* `all`, `basic`, `pro`\n\n"
            "*Example:*\n"
            "`/sendupdate updatekb 1.2.1 all Added 3 new KB articles`\n\n"
            "*Individual node:*\n"
            "`/sendupdate updatekb 1.2.1 all --email user@mail.com Summary`",
            parse_mode="Markdown",
        )
        return

    update_type = args[0]
    version = args[1]
    tier = args[2]
    remaining = args[3:]

    # Check for --node or --email flags
    target_node = None
    target_email = None
    summary_parts = []

    i = 0
    while i < len(remaining):
        if remaining[i] == "--node" and i + 1 < len(remaining):
            target_node = remaining[i + 1]
            i += 2
        elif remaining[i] == "--email" and i + 1 < len(remaining):
            target_email = remaining[i + 1]
            i += 2
        else:
            summary_parts.append(remaining[i])
            i += 1

    summary = " ".join(summary_parts)

    await update.message.reply_text("Creating update packet...")

    try:
        from tools.admin.update_packet import UpdatePacketBuilder, UPDATE_TYPES
        if update_type not in UPDATE_TYPES:
            await update.message.reply_text(
                f"Unknown type: `{update_type}`\nValid: {', '.join(UPDATE_TYPES.keys())}",
                parse_mode="Markdown",
            )
            return

        builder = UpdatePacketBuilder(
            update_type=update_type,
            tier=tier,
            version=version,
            target_node=target_node,
        )
        builder.set_summary(summary)

        # Check if there's a pending payload in context (set by /setpayload)
        payload = context.user_data.get("pending_update_payload", {})
        builder.set_payload(payload)

        pdf_path = builder.build()

        if target_email:
            # Send via email to individual node
            from services.email_service import EmailService
            svc = EmailService()
            # For email, we need to attach the PDF
            import smtplib
            from email.mime.multipart import MIMEMultipart
            from email.mime.text import MIMEText
            from email.mime.application import MIMEApplication

            msg = MIMEMultipart()
            msg["From"] = svc.from_email
            msg["To"] = target_email
            msg["Subject"] = f"BizNode Update Packet: {UPDATE_TYPES[update_type]} v{version}"
            msg.attach(MIMEText(
                f"BizNode Update Packet\n\n"
                f"Type: {update_type}\nVersion: {version}\nTier: {tier}\n\n"
                f"{summary}\n\nApply this update by placing the PDF in your node's inbox.",
                "plain",
            ))
            with open(pdf_path, "rb") as f:
                att = MIMEApplication(f.read(), _subtype="pdf")
                att.add_header("Content-Disposition", "attachment",
                               filename=os.path.basename(pdf_path))
                msg.attach(att)

            server = smtplib.SMTP(svc.smtp_host, svc.smtp_port)
            server.starttls()
            server.login(svc.smtp_user, svc.smtp_password)
            server.sendmail(svc.from_email, [target_email], msg.as_string())
            server.quit()

            await update.message.reply_text(
                f"Update packet sent to `{target_email}`\n"
                f"Type: {update_type} | Version: {version} | Tier: {tier}",
                parse_mode="Markdown",
            )
        else:
            # Send PDF to Telegram group (broadcast)
            if not TELEGRAM_GROUP_ID:
                # If no group, send to owner as file
                with open(pdf_path, "rb") as f:
                    await update.message.reply_document(
                        document=f,
                        filename=os.path.basename(pdf_path),
                        caption=f"Update Packet: {UPDATE_TYPES[update_type]} v{version} ({tier})\n{summary}",
                    )
                return

            caption = (
                f"*BizNode Update Packet*\n\n"
                f"Type: `{update_type}`\n"
                f"Version: `{version}`\n"
                f"Tier: `{tier}`\n\n"
                f"{summary}\n\n"
                f"_Nodes: download and apply this update automatically._"
            )
            with open(pdf_path, "rb") as f:
                await context.bot.send_document(
                    chat_id=TELEGRAM_GROUP_ID,
                    document=f,
                    filename=os.path.basename(pdf_path),
                    caption=caption,
                    parse_mode="Markdown",
                )
            await update.message.reply_text(
                f"Update packet posted to group.\n"
                f"Type: {update_type} | Version: {version} | Tier: {tier}",
            )

        # Clear pending payload
        context.user_data.pop("pending_update_payload", None)
        logger.info("Update packet sent: %s v%s (%s)", update_type, version, tier)

    except Exception as exc:
        logger.error("sendupdate failed: %s", exc, exc_info=True)
        await update.message.reply_text(f"Failed to create update packet: {exc}")


async def cmd_setpayload(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Set JSON payload for the next /sendupdate command.

    Usage: /setpayload {"articles": [{"title": "...", "content": "..."}]}
    """
    user_id = update.effective_user.id
    if not is_owner(user_id):
        await update.message.reply_text("Owner only.")
        return

    raw = " ".join(context.args) if context.args else ""
    if not raw.strip():
        await update.message.reply_text(
            "Usage: `/setpayload {\"articles\": [...]}`\n\n"
            "Set JSON payload, then run `/sendupdate` to build the PDF.",
            parse_mode="Markdown",
        )
        return

    try:
        payload = json.loads(raw)
        context.user_data["pending_update_payload"] = payload
        keys = list(payload.keys())
        await update.message.reply_text(
            f"Payload set. Keys: {keys}\nNow run `/sendupdate` to create the packet.",
            parse_mode="Markdown",
        )
    except json.JSONDecodeError as exc:
        await update.message.reply_text(f"Invalid JSON: {exc}")


async def handle_update_packet_document(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Process incoming PDF update packets sent to the bot."""
    msg = update.message
    if not msg or not msg.document:
        return

    doc = msg.document
    # Only process PDFs with our naming convention
    if not doc.file_name or not doc.file_name.startswith("packet_") or not doc.file_name.endswith(".pdf"):
        return

    user_id = update.effective_user.id
    if not is_owner(user_id):
        # Non-owners can receive updates but only if targeted to them
        pass

    await msg.reply_text("Processing update packet...")

    try:
        # Download the PDF
        import tempfile
        file = await doc.get_file()
        with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
            await file.download_to_drive(tmp.name)
            tmp_path = tmp.name

        from tools.admin.update_processor import process_packet, extract_metadata, BLOCKCHAIN_TYPES

        # First extract to check if it needs OTP
        metadata = extract_metadata(tmp_path)
        if metadata is None:
            await msg.reply_text("Invalid update packet: could not extract metadata.")
            os.unlink(tmp_path)
            return

        update_type = metadata.get("update_type", "")
        version = metadata.get("version", "")
        requires_otp = metadata.get("requires_otp", False)

        if requires_otp:
            # Store path for later OTP verification
            context.user_data["pending_otp_packet"] = tmp_path
            context.user_data["pending_otp_metadata"] = metadata

            # Send OTP to owner
            owner_email = os.getenv("OWNER_EMAIL", "")
            if owner_email:
                import random
                otp = str(random.randint(100000, 999999))
                context.user_data["packet_otp"] = otp

                from services.email_service import EmailService
                svc = EmailService()
                svc.send_email(
                    owner_email,
                    f"BizNode Update OTP: {update_type} v{version}",
                    f"Your OTP for applying {update_type} v{version}: {otp}\n\n"
                    f"Reply with /applyupdate {otp} to confirm.",
                )

            await msg.reply_text(
                f"*Blockchain Update Requires OTP*\n\n"
                f"Type: `{update_type}`\nVersion: `{version}`\n\n"
                f"OTP sent to owner email.\n"
                f"Reply: `/applyupdate <OTP>` to apply.",
                parse_mode="Markdown",
            )
            return

        # Non-blockchain: process immediately
        node_tier = os.getenv("NODE_TIER", "basic")
        result = process_packet(tmp_path, node_tier=node_tier)

        status = "SUCCESS" if result.success else "FAILED"
        await msg.reply_text(
            f"*Update Packet {status}*\n\n"
            f"Type: `{result.update_type}`\n"
            f"Version: `{result.version}`\n"
            f"Message: {result.message}",
            parse_mode="Markdown",
        )

        os.unlink(tmp_path)
        logger.info("Update packet processed: %s v%s -> %s",
                     result.update_type, result.version, status)

    except Exception as exc:
        logger.error("Update packet processing failed: %s", exc, exc_info=True)
        await msg.reply_text(f"Error processing update packet: {exc}")


async def cmd_applyupdate(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Apply a pending blockchain update after OTP verification.
    Usage: /applyupdate <OTP>
    """
    user_id = update.effective_user.id
    if not is_owner(user_id):
        await update.message.reply_text("Owner only.")
        return

    otp_input = (context.args[0] if context.args else "").strip()
    stored_otp = context.user_data.get("packet_otp", "")
    pending_path = context.user_data.get("pending_otp_packet", "")

    if not pending_path:
        await update.message.reply_text("No pending update packet. Send a PDF first.")
        return

    if not otp_input or otp_input != stored_otp:
        await update.message.reply_text("Invalid OTP. Check your email and try again.")
        return

    await update.message.reply_text("OTP verified. Applying blockchain update...")

    try:
        from tools.admin.update_processor import process_packet
        node_tier = os.getenv("NODE_TIER", "pro")
        result = process_packet(pending_path, node_tier=node_tier, otp_verified=True)

        status = "SUCCESS" if result.success else "FAILED"
        await update.message.reply_text(
            f"*Blockchain Update {status}*\n\n"
            f"Type: `{result.update_type}`\n"
            f"Version: `{result.version}`\n"
            f"Message: {result.message}",
            parse_mode="Markdown",
        )

        # Cleanup
        if os.path.exists(pending_path):
            os.unlink(pending_path)
        context.user_data.pop("pending_otp_packet", None)
        context.user_data.pop("pending_otp_metadata", None)
        context.user_data.pop("packet_otp", None)

    except Exception as exc:
        logger.error("Apply update failed: %s", exc)
        await update.message.reply_text(f"Failed: {exc}")


# ---------------------------------------------------------------------------
# /bo — Ad Marketplace commands
# ---------------------------------------------------------------------------

async def cmd_bo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Ad Marketplace command hub.

    Usage:
      /bo                         — Show marketplace help
      /bo submit <title>          — Submit ad proposal
      /bo confirm <id> <otp>      — Confirm proposal with OTP
      /bo quote <id> <amount>     — Submit quote for a proposal
      /bo accept <id>             — Accept consolidated quote
      /bo status <id>             — Check proposal/campaign status
      /bo campaigns               — List active campaigns
      /bo stats                   — Marketplace statistics
    """
    user_id = update.effective_user.id
    args = context.args or []
    subcmd = args[0].lower() if args else ""

    if not subcmd:
        help_text = (
            "*Ad Marketplace (/bo)*\n\n"
            "Decentralized advertising powered by DZIT credits & BZeUSD tokens.\n\n"
            "*Commands:*\n"
            "`/bo submit <title>` - Submit ad proposal\n"
            "`/bo confirm <id> <otp>` - Confirm with OTP\n"
            "`/bo quote <id> <amount>` - Quote for a proposal\n"
            "`/bo accept <id>` - Accept consolidated quote\n"
            "`/bo status <id>` - Check status\n"
            "`/bo campaigns` - Active campaigns\n"
            "`/bo stats` - Marketplace stats\n\n"
            "*Flow:* Submit -> OTP -> Broadcast -> Quote -> Evaluate -> Publish -> Monitor\n"
            "*Fee:* 1 DZIT per proposal (deducted after OTP confirmation)\n"
            "*Incentive:* 50% BZeUSD + 50% DZIT credits"
        )
        await update.message.reply_text(help_text, parse_mode="Markdown")
        return

    if subcmd == "submit":
        await _bo_submit(update, context, args[1:])
    elif subcmd == "confirm":
        await _bo_confirm(update, context, args[1:])
    elif subcmd == "quote":
        await _bo_quote(update, context, args[1:])
    elif subcmd == "accept":
        await _bo_accept(update, context, args[1:])
    elif subcmd == "status":
        await _bo_status(update, context, args[1:])
    elif subcmd == "campaigns":
        await _bo_campaigns(update, context)
    elif subcmd == "stats":
        await _bo_stats(update, context)
    else:
        await update.message.reply_text(
            f"Unknown subcommand: `{subcmd}`\nUse `/bo` for help.", parse_mode="Markdown"
        )


async def _bo_submit(update, context, args):
    """Submit a new ad proposal."""
    user_id = update.effective_user.id
    if not args:
        await update.message.reply_text(
            "Usage: `/bo submit <campaign title>`\n"
            "Example: `/bo submit DeFi Token Launch Promotion`",
            parse_mode="Markdown",
        )
        return

    title = " ".join(args)

    # Get user email from bot_users
    try:
        r = requests.get(
            f"{DASHBOARD_URL}/api/users/{user_id}",
            headers={"X-Bot-Secret": BOT_SECRET},
            timeout=5,
        )
        if r.status_code != 200:
            await update.message.reply_text("You must be registered first. Send /start to begin onboarding.")
            return
        user_data = r.json()
        email = user_data.get("email", "")
        if not email:
            await update.message.reply_text("No email registered. Complete onboarding with /start first.")
            return
    except Exception as exc:
        logger.error("Failed to get user data: %s", exc)
        await update.message.reply_text("Could not verify your registration. Try again later.")
        return

    # Submit proposal via API
    try:
        r = requests.post(
            f"{DASHBOARD_URL}/api/bizchannel/marketplace/proposals",
            json={
                "advertiser_email": email,
                "advertiser_telegram_id": str(user_id),
                "campaign_title": title,
                "business_category": "general",
            },
            headers={"X-Bot-Secret": BOT_SECRET},
            timeout=10,
        )
        data = r.json()
        if data.get("ok"):
            pid = data.get("proposal_id")
            await update.message.reply_text(
                f"*Proposal #{pid} Submitted*\n\n"
                f"Title: {title}\n"
                f"An OTP has been sent to your registered email.\n\n"
                f"Confirm within 24 hours:\n"
                f"`/bo confirm {pid} <OTP>`\n\n"
                f"1 DZIT will be deducted upon confirmation.",
                parse_mode="Markdown",
            )
        else:
            await update.message.reply_text(f"Submission failed: {data}")
    except Exception as exc:
        logger.error("Proposal submit failed: %s", exc)
        await update.message.reply_text(f"Error: {exc}")


async def _bo_confirm(update, context, args):
    """Confirm a proposal with OTP."""
    if len(args) < 2:
        await update.message.reply_text(
            "Usage: `/bo confirm <proposal_id> <otp>`", parse_mode="Markdown"
        )
        return

    proposal_id = args[0]
    otp = args[1]

    try:
        r = requests.post(
            f"{DASHBOARD_URL}/api/bizchannel/marketplace/proposals/{proposal_id}/confirm",
            json={"otp": otp},
            headers={"X-Bot-Secret": BOT_SECRET},
            timeout=10,
        )
        data = r.json()
        if r.status_code == 200 and data.get("ok"):
            await update.message.reply_text(
                f"*Proposal #{proposal_id} Confirmed*\n\n"
                f"1 DZIT deducted.\n"
                f"Your proposal is being broadcast to matching nodes.\n"
                f"Node owners have 24 hours to submit quotes.",
                parse_mode="Markdown",
            )
        else:
            msg = data.get("detail", data.get("message", str(data)))
            await update.message.reply_text(f"Confirmation failed: {msg}")
    except Exception as exc:
        logger.error("Proposal confirm failed: %s", exc)
        await update.message.reply_text(f"Error: {exc}")


async def _bo_quote(update, context, args):
    """Submit a quote for a proposal."""
    if len(args) < 2:
        await update.message.reply_text(
            "Usage: `/bo quote <proposal_id> <amount>`\n"
            "Example: `/bo quote 5 100`",
            parse_mode="Markdown",
        )
        return

    proposal_id = args[0]
    try:
        amount = float(args[1])
    except ValueError:
        await update.message.reply_text("Amount must be a number.")
        return

    user_id = update.effective_user.id

    # Get user info
    try:
        r = requests.get(
            f"{DASHBOARD_URL}/api/users/{user_id}",
            headers={"X-Bot-Secret": BOT_SECRET},
            timeout=5,
        )
        if r.status_code != 200:
            await update.message.reply_text("You must be registered. Use /start first.")
            return
        user_data = r.json()
        email = user_data.get("email", "")
    except Exception:
        await update.message.reply_text("Could not verify registration.")
        return

    # Submit quote via API
    try:
        r = requests.post(
            f"{DASHBOARD_URL}/api/bizchannel/marketplace/quotes",
            json={
                "proposal_id": int(proposal_id),
                "node_owner_email": email,
                "node_owner_telegram_id": str(user_id),
                "quoted_amount": amount,
                "node_tier": os.getenv("NODE_TIER", "basic"),
            },
            headers={"X-Bot-Secret": BOT_SECRET},
            timeout=10,
        )
        data = r.json()
        if data.get("ok"):
            qid = data.get("quote_id")
            await update.message.reply_text(
                f"*Quote #{qid} Submitted*\n\n"
                f"Proposal: #{proposal_id}\n"
                f"Amount: {amount} BZeUSD/DZIT\n\n"
                f"Harmonious scoring will evaluate your group activity.\n"
                f"Payout: 50% BZeUSD + 50% DZIT credits.",
                parse_mode="Markdown",
            )
        else:
            msg = data.get("detail", str(data))
            await update.message.reply_text(f"Quote failed: {msg}")
    except Exception as exc:
        logger.error("Quote submit failed: %s", exc)
        await update.message.reply_text(f"Error: {exc}")


async def _bo_accept(update, context, args):
    """Accept the consolidated quote for a proposal."""
    if not args:
        await update.message.reply_text(
            "Usage: `/bo accept <proposal_id>`", parse_mode="Markdown"
        )
        return

    proposal_id = args[0]
    user_id = update.effective_user.id

    try:
        r = requests.post(
            f"{DASHBOARD_URL}/api/bizchannel/marketplace/proposals/{proposal_id}/accept",
            json={"advertiser_telegram_id": str(user_id)},
            headers={"X-Bot-Secret": BOT_SECRET},
            timeout=10,
        )
        data = r.json()
        if r.status_code == 200 and data.get("ok"):
            consolidated = data.get("consolidated", {})
            await update.message.reply_text(
                f"*Proposal #{proposal_id} Accepted*\n\n"
                f"Accepted quotes: {consolidated.get('accepted_quotes', 0)}\n"
                f"Total BZeUSD: {consolidated.get('total_bzeusd_payout', 0)}\n"
                f"Total DZIT: {consolidated.get('total_dzit_payout', 0)}\n"
                f"Total cost: {consolidated.get('total_cost', 0)}\n\n"
                f"Campaign publishing initiated.\n"
                f"15-day monitoring starts now.",
                parse_mode="Markdown",
            )
        else:
            msg = data.get("detail", str(data))
            await update.message.reply_text(f"Accept failed: {msg}")
    except Exception as exc:
        logger.error("Proposal accept failed: %s", exc)
        await update.message.reply_text(f"Error: {exc}")


async def _bo_status(update, context, args):
    """Check proposal or campaign status."""
    if not args:
        await update.message.reply_text(
            "Usage: `/bo status <proposal_id>`", parse_mode="Markdown"
        )
        return

    proposal_id = args[0]
    try:
        r = requests.get(
            f"{DASHBOARD_URL}/api/bizchannel/marketplace/proposals/{proposal_id}",
            headers={"X-Bot-Secret": BOT_SECRET},
            timeout=10,
        )
        if r.status_code == 404:
            await update.message.reply_text(f"Proposal #{proposal_id} not found.")
            return
        data = r.json()
        quotes = data.get("quotes", [])
        campaign = data.get("campaign")

        lines = [
            f"*Proposal #{proposal_id}*\n",
            f"Title: {data.get('campaign_title', 'N/A')}",
            f"Status: `{data.get('status', 'unknown')}`",
            f"Category: {data.get('business_category', 'N/A')}",
            f"DZIT deducted: {'Yes' if data.get('dzit_deducted') else 'No'}",
            f"\n*Quotes:* {len(quotes)}",
        ]

        for q in quotes[:5]:
            score = q.get("harmonious_score", 0)
            lines.append(
                f"  #{q['id']}: {q.get('quoted_amount', 0)} -> score {score:.1f} "
                f"({q.get('status', '')})"
            )

        if campaign:
            lines.append(f"\n*Campaign:* {campaign.get('status', '')}")
            lines.append(f"Monitoring: {campaign.get('monitoring_status', 'N/A')}")
            lines.append(f"BZeUSD paid: {campaign.get('total_payout_bzeusd', 0)}")
            lines.append(f"DZIT paid: {campaign.get('total_payout_dzit', 0)}")

        await update.message.reply_text("\n".join(lines), parse_mode="Markdown")
    except Exception as exc:
        logger.error("Status check failed: %s", exc)
        await update.message.reply_text(f"Error: {exc}")


async def _bo_campaigns(update, context):
    """List active campaigns."""
    try:
        r = requests.get(
            f"{DASHBOARD_URL}/api/bizchannel/marketplace/campaigns?status=active&limit=10",
            headers={"X-Bot-Secret": BOT_SECRET},
            timeout=10,
        )
        campaigns = r.json()
        if not campaigns:
            await update.message.reply_text("No active campaigns.")
            return

        lines = ["*Active Ad Campaigns*\n"]
        for c in campaigns:
            lines.append(
                f"Campaign #{c['id']} (Proposal #{c.get('proposal_id', '?')})\n"
                f"  Status: {c.get('status', '')} | Monitoring: {c.get('monitoring_status', '')}\n"
                f"  Quotes: {c.get('total_accepted_quotes', 0)} | "
                f"BZeUSD: {c.get('total_payout_bzeusd', 0)} | DZIT: {c.get('total_payout_dzit', 0)}"
            )
        await update.message.reply_text("\n".join(lines), parse_mode="Markdown")
    except Exception as exc:
        await update.message.reply_text(f"Error: {exc}")


async def _bo_stats(update, context):
    """Show marketplace statistics."""
    try:
        r = requests.get(
            f"{DASHBOARD_URL}/api/bizchannel/marketplace/stats",
            headers={"X-Bot-Secret": BOT_SECRET},
            timeout=10,
        )
        s = r.json()
        await update.message.reply_text(
            f"*Ad Marketplace Stats*\n\n"
            f"Proposals: {s.get('total_proposals', 0)} "
            f"(published: {s.get('published_proposals', 0)})\n"
            f"Quotes: {s.get('total_quotes', 0)} "
            f"(accepted: {s.get('accepted_quotes', 0)})\n"
            f"Campaigns: {s.get('total_campaigns', 0)} "
            f"(active: {s.get('active_campaigns', 0)})\n"
            f"BZeUSD paid: {s.get('total_bzeusd_paid', 0)}\n"
            f"DZIT paid: {s.get('total_dzit_paid', 0)}\n"
            f"Avg harmonious score: {s.get('avg_harmonious_score', 0)}",
            parse_mode="Markdown",
        )
    except Exception as exc:
        await update.message.reply_text(f"Error: {exc}")


# ---------------------------------------------------------------------------
# Group message handler
# ---------------------------------------------------------------------------

async def handle_group_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Silently capture leads from group messages; respond only when @mentioned."""
    if not TELEGRAM_GROUP_ID:
        return

    msg      = update.message
    if not msg or not msg.text:
        return

    user     = update.effective_user
    user_id  = user.id
    username = user.username or ""
    name     = (user.first_name or "") + (" " + user.last_name if user.last_name else "")
    text     = msg.text.strip()

    # Silently capture as lead
    upsert_group_lead(name.strip(), user_id, username)
    logger.info("GROUP member captured: telegram_id=%s username=%s", user_id, username)

    # Respond if bot is @mentioned
    bot_username = context.bot.username or ""
    if bot_username and f"@{bot_username}" in text:
        query = text.replace(f"@{bot_username}", "").strip()
        if query:
            thinking = await msg.reply_text("⏳ Thinking...")
            loop = asyncio.get_event_loop()
            response = await loop.run_in_executor(None, rag_answer, user_id, query, username)
            await thinking.edit_text(response)


# ---------------------------------------------------------------------------
# /help intake handler — processes 4-step questionnaire responses
# ---------------------------------------------------------------------------

async def _handle_help_intake(update: Update, user_id: int, text: str) -> bool:
    """
    Process /help intake steps. Returns True if we handled it (consume the message).
    After all 4 steps, generates BizBrain profile and optionally emails it.
    """
    intake = _help_intake.get(user_id)
    if not intake:
        return False

    step = intake.get("step", 1)

    if step == 1:
        intake["business_type"] = text
        intake["step"] = 2
        await update.message.reply_text(
            f"*Question 2 of 4:* {HELP_STEP_QUESTIONS[2]}",
            parse_mode="Markdown",
        )
        return True

    if step == 2:
        intake["services"] = text
        intake["step"] = 3
        await update.message.reply_text(
            f"*Question 3 of 4:* {HELP_STEP_QUESTIONS[3]}",
            parse_mode="Markdown",
        )
        return True

    if step == 3:
        intake["target_audience"] = text
        intake["step"] = 4
        await update.message.reply_text(
            f"*Question 4 of 4:* {HELP_STEP_QUESTIONS[4]}",
            parse_mode="Markdown",
        )
        return True

    if step == 4:
        intake["challenges"] = text
        # Remove from in-progress dict
        _help_intake.pop(user_id, None)

        await update.message.reply_text(
            "Generating your BizBrain profile... this takes 20-30 seconds.",
            parse_mode="Markdown",
        )

        # Generate profile in background using executor
        import asyncio
        loop = asyncio.get_event_loop()

        try:
            from services.bizbrain_service import (
                check_website_registration,
                generate_bizbrain_profile,
                send_bizbrain_email,
            )
        except ImportError as e:
            await update.message.reply_text(f"BizBrain service unavailable: {e}")
            return True

        profile_md = await loop.run_in_executor(None, generate_bizbrain_profile, intake)

        # Check if they have a purchased license
        user_rec  = get_bot_user(user_id)
        email     = user_rec.get("email", "")
        profile   = get_profile()
        biz_name  = (profile.get("business_name") or "your business").strip()
        persona   = (profile.get("ai_persona_name") or "BizNode").strip()

        if email:
            is_registered = await loop.run_in_executor(
                None, check_website_registration, email
            )
        else:
            is_registered = False

        if is_registered and email:
            # Send full profile by email
            await loop.run_in_executor(
                None, send_bizbrain_email, email, profile_md, biz_name, persona
            )
            # Show preview
            preview = profile_md[:800] + ("..." if len(profile_md) > 800 else "")
            await update.message.reply_text(
                f"*Your BizBrain Profile (Preview)*\n\n{preview}\n\n"
                f"The full profile has been sent to *{email}*.",
                parse_mode="Markdown",
            )
        else:
            # Show full profile in chat + upsell
            preview = profile_md[:1500] + ("..." if len(profile_md) > 1500 else "")
            group_note = f"\n\nJoin our community: {TELEGRAM_GROUP_LINK}" if TELEGRAM_GROUP_LINK else ""
            await update.message.reply_text(
                f"*Your BizBrain Profile*\n\n{preview}\n\n"
                f"Purchase a BizNode license at https://1bz.biz to receive the full profile "
                f"by email and unlock all automation features.{group_note}",
                parse_mode="Markdown",
            )
        return True

    return False


# ---------------------------------------------------------------------------
# Main message handler
# ---------------------------------------------------------------------------

# ---------------------------------------------------------------------------
# Owner action executor — parse owner instructions as actions, not questions
# ---------------------------------------------------------------------------
_PENDING_OWNER_CHANGES: dict[int, dict] = {}  # telegram_id → {type, value}


def _mask_email(email: str) -> str:
    """Mask email for display: c***@1bz.biz"""
    if "@" not in email:
        return "***"
    local, domain = email.split("@", 1)
    if len(local) <= 1:
        return f"*@{domain}"
    return f"{local[0]}***@{domain}"


# Strict format: send email to X subject Y body Z
# Captures email with optional spaces around @ (e.g., "Shashi @1bz.biz")
_EMAIL_ACTION_RE = re.compile(
    r"send\s+(?:test\s+)?email\s+to\s+(\S+\s*@\s*\S+)"
    r"(?:\s+subject\s+(.+?))?(?:\s+body\s+(.+))?$",
    re.IGNORECASE,
)
# Natural language: "send email to X about Y", "mail X about Y", "email X regarding Y"
# Handles spaces around @ symbol
_EMAIL_NATURAL_RE = re.compile(
    r"(?:send\s+(?:an?\s+)?(?:email|mail)|mail|email)\s+(?:to\s+)?(\S+\s*@\s*\S+)"
    r"\s+(?:about|regarding|on|re|explaining|with)\s+(.+)$",
    re.IGNORECASE,
)
_BLOCK_ACTION_RE  = re.compile(r"block\s+(?:user\s+)?(\d+)", re.IGNORECASE)
_UNBLOCK_ACTION_RE = re.compile(r"unblock\s+(?:user\s+)?(\d+)", re.IGNORECASE)


def _normalize_email(raw: str, context: str = "") -> str:
    """Fix common email typos: collapse spaces around @, lowercase.

    Handles inputs like 'Shashi @1bz.biz', 'user @ domain.com',
    or even 'Shashi' when context contains '@domain'.
    """
    # If raw already has @ (possibly with spaces), collapse
    if "@" in raw:
        return re.sub(r"\s*@\s*", "@", raw).strip().lower()
    # raw has no @ — check if next word in context starts with @
    if context:
        m = re.search(re.escape(raw) + r"\s+(@[^\s]+)", context, re.IGNORECASE)
        if m:
            return (raw + m.group(1)).strip().lower()
    return raw.strip().lower()


def _execute_owner_action(text: str, user_id: int = 0) -> str | None:
    """Try to parse owner text as an actionable instruction.

    Returns a response string if an action was executed, or None to fall
    through to the normal RAG pipeline.
    """
    # ── Send email (strict format: subject + body given) ────────────
    m = _EMAIL_ACTION_RE.search(text)
    if m and m.group(2) and m.group(3):
        to_email = _normalize_email(m.group(1), text)
        subject  = m.group(2).strip()
        body     = m.group(3).strip()
        try:
            from services.email_service import EmailService
            svc = EmailService()
            svc.send_email(to_email, subject, body)
            return f"✅ Email sent to {to_email}\nSubject: {subject}"
        except Exception as exc:
            return f"Failed to send email: {exc}"

    # ── Send email (natural language — AI composes the email) ─────
    m_natural = _EMAIL_NATURAL_RE.search(text)
    if not m_natural:
        # Fallback: strict regex matched email but no body — treat as natural
        m_strict = _EMAIL_ACTION_RE.search(text)
        if m_strict:
            m_natural = m_strict
    if m_natural:
        to_email = _normalize_email(m_natural.group(1), text)
        # Extract topic from whatever follows the email address
        topic = m_natural.group(2).strip() if m_natural.lastindex >= 2 and m_natural.group(2) else "general inquiry"
        # Use LLM to compose a professional email
        compose_prompt = (
            f"Compose a professional SALES email about: {topic}\n"
            f"Recipient: {to_email}\n"
            f"Sender: The business owner of 1BZ DZIT DAO LLC\n\n"
            f"IMPORTANT: If the topic mentions BizNode or explaining/introducing BizNode, write a SALES email.\n"
            f"NEVER write a partnership or collaboration email unless the topic explicitly says 'partnership'.\n"
            f"Sales emails must highlight:\n"
            f"- What BizNode does: AI-powered business automation (chatbot, CRM, lead capture, support KB, email automation, scheduling, analytics)\n"
            f"- Pricing: Basic $200 (8 core features), Pro $499 (16+ features including BizChannel, payments, API access)\n"
            f"- 10-day free trial available\n"
            f"- ROI: runs 24/7, captures leads automatically, no cloud dependency\n"
            f"- CTA: visit https://biznode.1bz.biz or contact @biznode_bot on Telegram\n"
            f"- Sign off as the business owner (1BZ DZIT DAO LLC), not 'BizNode Team'\n\n"
            f"Write ONLY the email subject on the first line prefixed with 'Subject: ', "
            f"then a blank line, then the email body. Keep it persuasive, professional, and concise. "
            f"Use knowledge base context if relevant."
        )
        try:
            ai_response = query_llm(compose_prompt, owner_mode=True)
            # Parse subject and body from AI response
            lines = ai_response.strip().split("\n", 1)
            if lines[0].lower().startswith("subject:"):
                subject = lines[0].split(":", 1)[1].strip()
                body = lines[1].strip() if len(lines) > 1 else topic
            else:
                subject = topic.title()
                body = ai_response.strip()

            from services.email_service import EmailService
            svc = EmailService()
            svc.send_email(to_email, subject, body)
            return (
                f"✅ Email sent to {to_email}\n\n"
                f"*Subject:* {subject}\n\n"
                f"*Body:*\n{body[:500]}"
            )
        except Exception as exc:
            return f"Failed to compose/send email: {exc}"

    # ── Block user ────────────────────────────────────────────────────
    m = _BLOCK_ACTION_RE.search(text)
    if m:
        tid = m.group(1)
        try:
            r = requests.post(
                f"{DASHBOARD_URL}/api/customers/blocked/{tid}",
                headers=DASH_HEADERS, timeout=5,
            )
            if r.status_code < 300:
                return f"User {tid} has been blocked."
            return f"Block failed: {r.text}"
        except Exception as exc:
            return f"Block failed: {exc}"

    # ── Unblock user ──────────────────────────────────────────────────
    m = _UNBLOCK_ACTION_RE.search(text)
    if m:
        tid = m.group(1)
        try:
            r = requests.delete(
                f"{DASHBOARD_URL}/api/customers/blocked/{tid}",
                headers=DASH_HEADERS, timeout=5,
            )
            if r.status_code < 300:
                return f"User {tid} has been unblocked."
            return f"Unblock failed: {r.text}"
        except Exception as exc:
            return f"Unblock failed: {exc}"

    # ── Update bot email (AGENT_EMAIL) — requires OTP ───────────────
    m = re.search(r"\b(?:bot|agent|biznode)\s+email\s+(?:is|:)\s*([^\s,]+@[^\s,]+)", text, re.I)
    if m:
        new_email = m.group(1).strip()
        owner_email = os.environ.get("OWNER_EMAIL", "") or os.environ.get("AGENT_EMAIL", "")
        if not owner_email:
            return "No owner email configured to send OTP. Set OWNER_EMAIL in dashboard settings first."
        _PENDING_OWNER_CHANGES[user_id] = {"type": "bot_email", "value": new_email}
        name = "BizNode Owner"
        if send_otp(user_id, owner_email, name):
            return f"OTP sent to *{_mask_email(owner_email)}* to verify bot email change to *{new_email}*.\n\nReply with the OTP code to confirm."
        return "Failed to send OTP. Check SMTP settings."

    # ── Update owner email (OWNER_EMAIL) — requires OTP ──────────────
    m = re.search(r"\b(?:my|owner)\s+email\s+(?:is|:)\s*([^\s,]+@[^\s,]+)", text, re.I)
    if m:
        new_email = m.group(1).strip()
        # Send OTP to the NEW email to verify ownership
        _PENDING_OWNER_CHANGES[user_id] = {"type": "owner_email", "value": new_email}
        name = "BizNode Owner"
        if send_otp(user_id, new_email, name):
            return f"OTP sent to *{_mask_email(new_email)}* to verify your new owner email.\n\nReply with the OTP code to confirm."
        return "Failed to send OTP. Check SMTP settings."

    # ── Update phone — requires OTP ───────────────────────────────────
    m = re.search(r"\b(?:my\s+)?phone\s+(?:is|:)\s*([+\d][\d\s\-]{6,})", text, re.I)
    if m:
        new_phone = m.group(1).strip()
        owner_email = os.environ.get("OWNER_EMAIL", "") or os.environ.get("AGENT_EMAIL", "")
        if not owner_email:
            return "No owner email configured to send OTP. Set OWNER_EMAIL in dashboard settings first."
        _PENDING_OWNER_CHANGES[user_id] = {"type": "phone", "value": new_phone}
        name = "BizNode Owner"
        if send_otp(user_id, owner_email, name):
            return f"OTP sent to *{_mask_email(owner_email)}* to verify phone change to *{new_phone}*.\n\nReply with the OTP code to confirm."
        return "Failed to send OTP. Check SMTP settings."

    # ── Verify OTP for pending settings change ────────────────────────
    if user_id in _PENDING_OWNER_CHANGES and re.match(r"^\d{4,6}$", text.strip()):
        pending = _PENDING_OWNER_CHANGES[user_id]
        verified, msg = verify_otp(user_id, text.strip())
        if verified:
            change_type = pending["type"]
            value = pending["value"]
            del _PENDING_OWNER_CHANGES[user_id]
            try:
                if change_type == "bot_email":
                    r = requests.post(
                        f"{DASHBOARD_URL}/api/settings",
                        json={"AGENT_EMAIL": value, "SMTP_USER": value},
                        headers=DASH_HEADERS, timeout=5,
                    )
                    if r.status_code < 300:
                        os.environ["AGENT_EMAIL"] = value
                        return f"Bot email verified and updated to *{value}*"
                elif change_type == "owner_email":
                    r = requests.post(
                        f"{DASHBOARD_URL}/api/settings",
                        json={"OWNER_EMAIL": value},
                        headers=DASH_HEADERS, timeout=5,
                    )
                    if r.status_code < 300:
                        os.environ["OWNER_EMAIL"] = value
                        return f"Owner email verified and updated to *{value}*"
                elif change_type == "phone":
                    r = requests.post(
                        f"{DASHBOARD_URL}/api/settings",
                        json={"BUSINESS_PHONE": value},
                        headers=DASH_HEADERS, timeout=5,
                    )
                    if r.status_code < 300:
                        os.environ["BUSINESS_PHONE"] = value
                        return f"Phone verified and updated to *{value}*"
                return f"Failed to save {change_type} update."
            except Exception as exc:
                return f"Failed to save update: {exc}"
        else:
            return f"OTP verification failed: {msg}. Try again or resend by repeating the command."

    # ── Profile / my profile — check FIRST (before leads which matches "customer") ─
    if re.search(r"\b(my\s+profile|business\s+profile|who\s+am\s+i|my\s+account|my\s+business|show\s+profile)", text, re.I):
        profile = get_profile()
        report = get_report()
        total_users = report.get("total_bot_users", 0)
        total_leads = report.get("total_leads", 0)
        verified = report.get("verified_customers", 0)
        # Get email/phone from env (settings)
        owner_email = os.environ.get("OWNER_EMAIL", "") or os.environ.get("AGENT_EMAIL", "")
        bot_email = os.environ.get("AGENT_EMAIL", "")
        owner_phone = os.environ.get("BUSINESS_PHONE", "")
        if profile:
            name = profile.get("business_name", "Not set")
            desc = profile.get("business_description", "")[:300]
            services = profile.get("services_offered", "")[:200]
            return (
                f"*Your Business Profile*\n\n"
                f"Business: *{name}*\n"
                f"Owner Email: {_mask_email(owner_email) if owner_email else 'Not set'}\n"
                f"Bot Email: {_mask_email(bot_email) if bot_email else 'Not set'}\n"
                f"Phone: {owner_phone or 'Not set'}\n"
                f"Services: {services or 'Not set'}\n"
                f"Description: {desc or 'Not set'}\n\n"
                f"*Stats:* {total_leads} leads | {verified} verified | {total_users} bot users\n\n"
                f"_Update via: 'my email is ...' or 'bot email is ...' (OTP required)_"
            )
        return "Business profile not configured yet. Set it up in the dashboard."

    # ── Leads / new customers ─────────────────────────────────────────
    # Only match short, direct command-style requests (< 60 chars)
    # Longer conversational sentences go to LLM for intelligent response
    _leads_re = (
        r"^("
        r"new\s+(customer|lead|user|sign.?up)s?"
        r"|show\s+(lead|customer)s?"
        r"|list\s+(lead|customer)s?"
        r"|recent\s+(lead|customer)s?"
        r"|today.?s?\s+(lead|customer)s?"
        r"|how\s+many\s+(customer|lead|user)s?"
        r"|any\s+(new\s+)?(customer|lead)s?"
        r"|lead\s+(list|count|report)"
        r"|customer\s+(list|count|report)"
        r")$"
    )
    if len(text) < 60 and re.search(_leads_re, text.strip(), re.I):
        leads = get_recent_leads(10)
        report = get_report()
        new_today = report.get("new_leads_today", 0)
        total = report.get("total_leads", 0)
        total_users = report.get("total_bot_users", 0)
        if leads:
            lines = []
            for i, l in enumerate(leads, 1):
                name = l.get("name") or l.get("telegram_id") or "Unknown"
                email = l.get("email") or l.get("contact_info") or ""
                status = l.get("status") or "new"
                created = l.get("created_at", "")[:10] if l.get("created_at") else ""
                lines.append(f"{i}. *{name}* — {_mask_email(email) if email else ''} [{status}] {created}")
            return (
                f"*Today: {new_today} new leads* | Total leads: {total} | Bot users: {total_users}\n\n"
                f"*Recent Leads:*\n" + "\n".join(lines)
            )
        return f"No new leads yet today. Total bot users: {total_users}"

    # ── Status / server / uptime ──────────────────────────────────────
    # Tightened: requires server/system/bot context, not just "service"
    _status_re = (
        r"\b("
        r"(server|bot|system|qdrant|ollama|dashboard)\s+(status|health|running|down|up)"
        r"|is.*(server|bot|system).*(running|down|up|online|offline)"
        r"|(uptime|downtime|server\s+down|server\s+hours)"
        r"|check\s+(status|health|server)"
        r")\b"
    )
    if re.search(_status_re, text, re.I):
        # Ollama
        try:
            r = requests.get(OLLAMA_URL.replace("/api/generate", ""), timeout=3)
            ollama = "running" if r.status_code < 500 else "error"
        except Exception:
            ollama = "offline"
        # Qdrant
        try:
            r = requests.get(f"{QDRANT_URL}/collections/{QDRANT_COLLECTION}", timeout=3)
            pts = r.json().get("result", {}).get("points_count", "?")
            qdrant = f"running ({pts} KB entries)"
        except Exception:
            qdrant = "offline"
        report = get_report()
        q_today = report.get("queries_today", 0)
        u_today = report.get("unique_users_today", 0)
        return (
            f"*System Status*\n\n"
            f"Ollama: `{ollama}` (model: {OLLAMA_MODEL})\n"
            f"Qdrant: `{qdrant}`\n"
            f"Dashboard: `{DASHBOARD_URL}`\n\n"
            f"*Today:* {q_today} queries from {u_today} unique users"
        )

    # ── Report / summary / overview ───────────────────────────────────
    if re.search(r"\b(daily\s+report|show\s+report|give\s+report|business\s+summary|today.?s?\s+summary|overview|analytics|metric)", text, re.I):
        report = get_report()
        q_today = report.get("queries_today", 0)
        u_today = report.get("unique_users_today", 0)
        l_today = report.get("new_leads_today", 0)
        t_leads = report.get("total_leads", 0)
        verified = report.get("verified_customers", 0)
        t_users = report.get("total_bot_users", 0)
        return (
            f"*Daily Report*\n\n"
            f"Queries today: {q_today}\n"
            f"Unique users today: {u_today}\n"
            f"New leads today: {l_today}\n"
            f"Total leads: {t_leads}\n"
            f"Verified customers: {verified}\n"
            f"Total bot users: {t_users}"
        )

    # ── Recent queries / questions from customers ─────────────────────
    # Tightened: avoid matching "question" in general sentences like "I have a question"
    if re.search(r"\b(recent\s+quer|customer\s+quer|show\s+quer|what\s+.*asked|chat\s*log|recent\s+message|query\s+log)", text, re.I):
        rows = get_recent_queries(10)
        if rows:
            lines = []
            for i, q in enumerate(rows, 1):
                who = q.get("username") or q.get("telegram_id") or "?"
                query = (q.get("query") or "")[:60]
                lines.append(f"{i}. [{who}] {query}")
            return "*Recent Customer Queries*\n\n" + "\n".join(lines)
        return "No customer queries yet."

    # ── Menu — show owner menu (handles "menu", "show menu", etc.) ─────
    if re.search(r"^(menu|show\s+menu|main\s+menu|commands|help\s+menu)$", text.strip(), re.I):
        # Return special marker that handle_message will detect
        return "__SHOW_OWNER_MENU__"

    # ── Price / pricing queries ──────────────────────────────────────
    if re.search(r"\b(biznode\s+)?(basic|pro)\s*(license\s+)?(price|cost|how\s+much|pricing)\b", text, re.I):
        return (
            "*BizNode Pricing*\n\n"
            "*Basic License:* $200 USD or 200 BZeUSD (one-time)\n"
            "  1 bot, 8 core features, software key\n\n"
            "*Pro License:* $499 USD or 499 BZeUSD (one-time)\n"
            "  Up to 5 bots, all 27 features, USB hardware key\n\n"
            "Use /buy\\_license to purchase."
        )

    if re.search(r"\b(price|cost|how\s+much|pricing)\b.*\b(biznode|license|package|plan)\b", text, re.I):
        return (
            "*BizNode Pricing*\n\n"
            "*Basic License:* $200 USD or 200 BZeUSD (one-time)\n"
            "  1 bot, 8 core features, software key\n\n"
            "*Pro License:* $499 USD or 499 BZeUSD (one-time)\n"
            "  Up to 5 bots, all 27 features, USB hardware key\n\n"
            "Use /buy\\_license to purchase."
        )

    # ── /mr — Marketing Reports Package ──────────────────────────────
    # /mr or /mr <subcommand> — service description + payment workflow
    _mr_match = re.match(r"^/mr(?:\s+(.+))?$", text, re.I)
    if _mr_match:
        _mr_arg = (_mr_match.group(1) or "").strip()
        try:
            from tools.admin.datavault.datavault_service import (
                BIZNODE_WALLET, BIZNODE_EMAIL, ensure_default_packages,
                get_package, check_node_registration, create_package_order,
                process_dzit_payment, list_package_orders, get_package_order,
            )
            ensure_default_packages()

            # ── OWNER view ──────────────────────────────────────────
            if is_owner(user_id):
                # /mr status — list all orders
                if _mr_arg.lower() == "status":
                    all_orders = list_package_orders(package_id="mr-marketing-reports")
                    if not all_orders:
                        return "*Marketing Reports — Admin*\n\nNo orders yet."
                    pending = [o for o in all_orders if o.get("status") == "pending"]
                    confirmed = [o for o in all_orders if o.get("status") == "confirmed"]
                    revenue = sum(float(o.get("amount", 0)) for o in confirmed)
                    lines = [
                        f"*Marketing Reports — Admin*\n",
                        f"Total orders: {len(all_orders)}",
                        f"Pending: {len(pending)}",
                        f"Confirmed: {len(confirmed)}",
                        f"Revenue: {revenue:.2f} BZeUSD\n",
                    ]
                    for o in all_orders[:15]:
                        tgt = o.get("target_url", "—") or "—"
                        email = _mask_email(o.get("buyer_email", "")) if o.get("buyer_email") else "—"
                        lines.append(
                            f"`{o['order_id']}` | {o.get('status','?')} | {email} | {tgt}"
                        )
                    return "\n".join(lines)

                # /mr generate ORDER_ID — manually trigger report generation
                _gen_match = re.match(r"generate\s+(PKG-\S+)", _mr_arg, re.I)
                if _gen_match:
                    gen_oid = _gen_match.group(1)
                    gen_order = get_package_order(gen_oid)
                    if not gen_order:
                        return f"Order not found: {gen_oid}"
                    if not gen_order.get("target_url"):
                        return f"Order {gen_oid} has no target URL set. Ask the customer to send `/mr_url {gen_oid} <url>`."
                    return (
                        f"*Report generation triggered*\n\n"
                        f"Order: {gen_oid}\n"
                        f"Target: {gen_order['target_url']}\n"
                        f"Customer: {_mask_email(gen_order.get('buyer_email', ''))}\n\n"
                        f"Reports will be generated and delivered within 24-48 hours."
                    )

                # /mr (no args) — admin dashboard overview
                all_orders = list_package_orders(package_id="mr-marketing-reports")
                pending = [o for o in all_orders if o.get("status") == "pending"]
                confirmed = [o for o in all_orders if o.get("status") == "confirmed"]
                revenue = sum(float(o.get("amount", 0)) for o in confirmed)
                lines = [
                    f"*Marketing Reports — Admin Dashboard*\n",
                    f"Pending orders: {len(pending)}",
                    f"Completed orders: {len(confirmed)}",
                    f"Total revenue: {revenue:.2f} BZeUSD\n",
                    f"*Commands:*",
                    f"`/mr status` — view all orders",
                    f"`/mr generate PKG-XXXXX` — trigger report generation",
                ]
                if pending:
                    lines.append(f"\n*Recent pending:*")
                    for o in pending[:5]:
                        tgt = o.get("target_url", "—") or "—"
                        email = _mask_email(o.get("buyer_email", "")) if o.get("buyer_email") else "—"
                        lines.append(f"`{o['order_id']}` | {email} | {tgt}")
                return "\n".join(lines)

            # ── PUBLIC / NODE user view ─────────────────────────────
            # Check if this telegram user has a registered node
            is_node = False
            buyer_email = ""
            try:
                r = requests.get(
                    f"{DASHBOARD_URL}/api/customers/{user_id}",
                    headers=DASH_HEADERS, timeout=5,
                )
                if r.status_code == 200:
                    user_data = r.json()
                    buyer_email = user_data.get("email", "")
                    if buyer_email:
                        is_node = check_node_registration(buyer_email)
            except Exception:
                pass

            if is_node and _mr_arg and _mr_arg.startswith("http"):
                # Registered node with URL — charge 1 DZIT and auto-process
                target_url = _mr_arg
                dzit_result = process_dzit_payment(buyer_email, 1, f"MR package: {target_url}")
                if dzit_result.get("status") == "ok":
                    order = create_package_order(
                        "mr-marketing-reports",
                        buyer_email=buyer_email,
                        target_url=target_url,
                        buyer_telegram=str(user_id),
                        is_node=True,
                    )
                    return (
                        f"*Marketing Reports Package — Paid via DZIT*\n\n"
                        f"1 DZIT deducted (balance: {dzit_result.get('new_balance', '?')})\n"
                        f"Order: `{order['order_id'] if order else 'N/A'}`\n"
                        f"Target: {target_url}\n\n"
                        f"Your 8 reports are being generated. "
                        f"They will be sent to *{_mask_email(buyer_email)}* when ready."
                    )
                else:
                    err = dzit_result.get("error", "Unknown error")
                    return (
                        f"DZIT payment failed: {err}\n\n"
                        f"You can pay 10 BZeUSD instead. Send to:\n"
                        f"`{BIZNODE_WALLET}`\n\n"
                        f"Then verify with `/mr_verify PKG-ID POLYGONSCAN_URL`"
                    )

            # Show service description & payment instructions
            price_section = ""
            if is_node:
                price_section = (
                    f"*Node Operator Price:* 1 DZIT token\n"
                    f"Use: `/mr <company_url>` to auto-pay with DZIT\n\n"
                    f"*Or public price:* 10 BZeUSD (Polygon)\n"
                )
            else:
                price_section = f"*Price:* 10 BZeUSD (Polygon network)\n"

            return (
                f"*Marketing Reports Package*\n\n"
                f"Get all 8 AI-powered marketing reports for any company:\n\n"
                f"1. Prospect Analysis\n"
                f"2. Competitive Intelligence\n"
                f"3. Ideal Customer Profile\n"
                f"4. Cold Outreach Sequences\n"
                f"5. Objection Handling Playbook\n"
                f"6. Partnership Proposals\n"
                f"7. Growth Playbook\n"
                f"8. Free Listing Sites\n\n"
                f"{price_section}\n"
                f"*Free resources:*\n"
                f"[Download free templates](https://biznode.1bz.biz/marketing-templates.php)\n"
                f"[Tutorial: Marketing Reports](https://biznode.1bz.biz/tutorial-marketing-reports.php)\n\n"
                f"*How to order:*\n"
                f"1. Send 10 BZeUSD to:\n"
                f"   `{BIZNODE_WALLET}`\n"
                f"2. Verify payment:\n"
                f"   `/mr_verify PKG-ID POLYGONSCAN_URL`\n"
                f"3. Send your target company URL:\n"
                f"   `/mr_url PKG-ID https://example.com`\n\n"
                f"Reports will be generated and emailed within 24-48 hours.\n\n"
                f"_BizNode node operators pay only 1 DZIT token._"
            )
        except Exception as exc:
            return f"Marketing Reports error: {exc}"

    # ── /mr_verify <polygonscan_url> — verify package payment ────────
    _mr_verify_match = re.match(
        r"^/mr[_\s]?verify(?:\s+(PKG-\S+))?\s+(https?://\S+)",
        text, re.I
    )
    if _mr_verify_match:
        _verify_oid = (_mr_verify_match.group(1) or "").strip()
        poly_url = _mr_verify_match.group(2)
        try:
            from tools.admin.datavault.datavault_service import (
                confirm_package_order, get_package_order,
                create_package_order, check_node_registration,
                BIZNODE_WALLET,
            )

            # Verify it's a valid Polygonscan URL
            if "polygonscan.com" not in poly_url.lower():
                return "Please provide a valid Polygonscan transaction URL."

            # If order ID provided, confirm that specific order
            if _verify_oid:
                order = get_package_order(_verify_oid)
                if not order:
                    return f"Order not found: {_verify_oid}"
                if order["status"] == "confirmed":
                    return f"Order `{_verify_oid}` is already confirmed."

                if confirm_package_order(_verify_oid, payment_ref=poly_url, polygonscan_url=poly_url):
                    target = order.get("target_url", "")
                    if target:
                        resp = (
                            f"*Payment Verified!*\n\n"
                            f"Order: `{_verify_oid}`\n"
                            f"Target: {target}\n\n"
                            f"Your 8 marketing reports are being generated "
                            f"and will be sent to *{_mask_email(order.get('buyer_email', ''))}* "
                            f"within 24-48 hours."
                        )
                    else:
                        resp = (
                            f"*Payment Verified!*\n\n"
                            f"Order: `{_verify_oid}`\n\n"
                            f"Now send your target company URL:\n"
                            f"`/mr_url {_verify_oid} https://example.com`"
                        )
                    # Notify owner
                    if OWNER_ID and user_id != OWNER_ID:
                        try:
                            requests.post(
                                f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage",
                                json={
                                    "chat_id": OWNER_ID,
                                    "text": f"*New MR Payment Verified*\n\nOrder: `{_verify_oid}`\nPolygonscan: {poly_url}\nUser: {user_id}",
                                    "parse_mode": "Markdown",
                                },
                                timeout=5,
                            )
                        except Exception:
                            pass
                    return resp
                else:
                    return f"Failed to confirm order `{_verify_oid}`. It may have already been processed."

            # No order ID — create a new order from payment proof
            buyer_email = ""
            try:
                r = requests.get(
                    f"{DASHBOARD_URL}/api/customers/{user_id}",
                    headers=DASH_HEADERS, timeout=5,
                )
                if r.status_code == 200:
                    buyer_email = r.json().get("email", "")
            except Exception:
                pass

            order = create_package_order(
                "mr-marketing-reports",
                buyer_email=buyer_email,
                target_url="",
                buyer_telegram=str(user_id),
                is_node=False,
            )
            if not order:
                return "Failed to create order. Please try again or contact support."

            new_oid = order["order_id"]
            confirm_package_order(new_oid, payment_ref=poly_url, polygonscan_url=poly_url)

            # Notify owner
            if OWNER_ID and user_id != OWNER_ID:
                try:
                    requests.post(
                        f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage",
                        json={
                            "chat_id": OWNER_ID,
                            "text": f"*New MR Payment Verified*\n\nOrder: `{new_oid}`\nPolygonscan: {poly_url}\nUser: {user_id}",
                            "parse_mode": "Markdown",
                        },
                        timeout=5,
                    )
                except Exception:
                    pass

            return (
                f"*Payment Verified!*\n\n"
                f"Order: `{new_oid}`\n\n"
                f"Now send your target company URL:\n"
                f"`/mr_url {new_oid} https://example.com`"
            )
        except Exception as exc:
            return f"Verification error: {exc}"

    # ── /mr_url <order_id> <url> — associate target URL with order ───
    _mr_url_match = re.match(
        r"^/mr[_\s]?url\s+(PKG-\S+)\s+(https?://\S+)",
        text, re.I
    )
    if _mr_url_match:
        _url_oid = _mr_url_match.group(1)
        _target_url = _mr_url_match.group(2)
        try:
            from tools.admin.datavault.datavault_service import (
                get_package_order, update_order_target_url,
            )
            order = get_package_order(_url_oid)
            if not order:
                return f"Order not found: `{_url_oid}`"
            if order.get("buyer_telegram") != str(user_id) and not is_owner(user_id):
                return "You can only set the target URL for your own orders."

            if update_order_target_url(_url_oid, _target_url):
                masked_email = _mask_email(order.get("buyer_email", "")) if order.get("buyer_email") else "your email"
                # Notify owner
                if OWNER_ID and user_id != OWNER_ID:
                    try:
                        requests.post(
                            f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage",
                            json={
                                "chat_id": OWNER_ID,
                                "text": f"*MR Order Ready for Generation*\n\nOrder: `{_url_oid}`\nTarget: {_target_url}\nCustomer: {user_id}\n\nRun `/mr generate {_url_oid}` to start.",
                                "parse_mode": "Markdown",
                            },
                            timeout=5,
                        )
                    except Exception:
                        pass
                return (
                    f"*Order Confirmed!*\n\n"
                    f"Order: `{_url_oid}`\n"
                    f"Target: {_target_url}\n\n"
                    f"Reports will be generated and sent to *{masked_email}* "
                    f"within 24-48 hours."
                )
            else:
                return f"Failed to update target URL for order `{_url_oid}`."
        except Exception as exc:
            return f"URL update error: {exc}"

    # ── DataVault commands ────────────────────────────────────────────
    # /vault or "show vault" or "datavault" — vault status & stats
    if re.search(r"^/vault\b|^vault\s+(?:status|stats|info)|^datavault\b|^show\s+vault", text, re.I):
        try:
            from tools.admin.datavault.datavault_service import get_vault_stats, list_plugins
            stats = get_vault_stats()
            plugins = list_plugins()
            plugin_names = ", ".join(p["name"] for p in plugins) if plugins else "none"
            return (
                f"*DataVault Status*\n\n"
                f"Plugins: {stats['plugins']} ({plugin_names})\n"
                f"Reports: {stats['reports']}\n"
                f"Purchases: {stats['purchases']}\n"
                f"Revenue: {stats['total_revenue']} BZeUSD"
            )
        except Exception as exc:
            return f"DataVault error: {exc}"

    # /vault commands — list available plugin commands
    if re.search(r"^/vault\s+commands|^vault\s+commands|^show\s+(?:vault\s+)?commands", text, re.I):
        try:
            from tools.admin.datavault.datavault_service import list_available_commands
            cmds = list_available_commands()
            if not cmds:
                return "No DataVault commands available. Register a plugin first."
            lines = []
            for c in cmds:
                lines.append(f"/{c['command']} — {c['description']} ({c['plugin_name']})")
            return "*DataVault Commands*\n\n" + "\n".join(lines)
        except Exception as exc:
            return f"DataVault error: {exc}"

    # /vault reports — list generated reports
    if re.search(r"^/vault\s+reports|^vault\s+reports|^show\s+reports", text, re.I):
        try:
            from tools.admin.datavault.datavault_service import list_reports
            reports = list_reports(limit=10)
            if not reports:
                return "No reports in the vault yet. Generate one with a plugin command."
            lines = []
            for r in reports:
                price_str = f"{r['price']} {r['currency']}" if r['price'] > 0 else "FREE"
                lines.append(f"• {r['title']} [{price_str}] (ID: {r['report_id']})")
            return "*DataVault Reports*\n\n" + "\n".join(lines)
        except Exception as exc:
            return f"DataVault error: {exc}"

    # /salesreport <url>, /competitors <url>, /icp <desc>, etc. — generate via plugin
    _dv_cmd_match = re.match(
        r"^/?(salesreport|competitors|icp|outreach|objections|proposal|growth|listings"
        r"|prospect|competitive|comp|sr|idealcustomer|coldoutreach|partnership"
        r"|growthplan|freelistings|directories)\s*(.*)",
        text, re.I
    )
    if _dv_cmd_match:
        cmd_name = _dv_cmd_match.group(1).lower()
        cmd_args = _dv_cmd_match.group(2).strip()
        try:
            from tools.admin.datavault.report_generator import generate_report
            result = generate_report(
                command=cmd_name,
                args=cmd_args,
                creator_node="admin",
            )
            if result.get("success"):
                size_kb = round(result.get("file_size", 0) / 1024, 1)
                return (
                    f"*Report Generated*\n\n"
                    f"Title: {result['title']}\n"
                    f"Plugin: {result.get('plugin', '')}\n"
                    f"Size: {size_kb} KB\n"
                    f"Report ID: {result['report_id']}\n\n"
                    f"PDF: {result['pdf_path']}"
                )
            else:
                return f"Generation failed: {result.get('message', 'Unknown error')}"
        except Exception as exc:
            return f"DataVault error: {exc}"

    # ── Owner improvement/feedback instruction (bot monitor) ────────
    _feedback_re = re.search(
        r"^(?:improve|feedback|bot\s+should)\s*:\s*(.+)",
        text.strip(), re.I | re.DOTALL,
    )
    if _feedback_re:
        import json as _json
        from datetime import datetime as _dt
        instruction = _feedback_re.group(1).strip()
        try:
            _feedback_dir = Path(os.environ.get("ROOT_DIR", ".")) / "logs" / "bot_monitor"
            # Also try the known admin node path
            if not _feedback_dir.exists():
                _feedback_dir = Path(__file__).resolve().parent.parent.parent / "logs" / "bot_monitor"
            _feedback_dir.mkdir(parents=True, exist_ok=True)
            _feedback_file = _feedback_dir / "owner_feedback.jsonl"
            entry = {
                "timestamp": _dt.now().isoformat(),
                "instruction": instruction,
                "applied": False,
            }
            with open(_feedback_file, "a", encoding="utf-8") as f:
                f.write(_json.dumps(entry) + "\n")
            return (
                "Got it — your improvement instruction has been recorded "
                "and will be applied in the next monitor cycle."
            )
        except Exception as exc:
            return f"Failed to save feedback: {exc}"

    # ── Register agent ─────────────────────────────────────────────────
    m = re.search(r"\bregister\s+agent\s+(.+)", text, re.I)
    if m:
        agent_name = m.group(1).strip()
        try:
            owner_email = os.environ.get("OWNER_EMAIL", "") or os.environ.get("AGENT_EMAIL", "")
            biz_name = os.environ.get("BUSINESS_NAME", "BizNode Owner")
            r = requests.post(
                f"{DASHBOARD_URL}/api/agent-registry/register",
                json={
                    "agent_name": agent_name,
                    "owner_name": biz_name,
                    "owner_email": owner_email,
                    "owner_type": "individual",
                },
                headers=DASH_HEADERS, timeout=10,
            )
            data = r.json()
            if data.get("success"):
                return (
                    f"*Agent Registered*\n\n"
                    f"Name: *{agent_name}*\n"
                    f"ID: `{data['agent_id']}`\n"
                    f"Status: Unverified\n\n"
                    f"A verification request has been sent to the CA (ca@1bz.biz).\n"
                    f"Registry: https://registry.1bz.biz/"
                )
            return f"Registration failed: {data.get('error', 'Unknown error')}"
        except Exception as exc:
            return f"Failed to register agent: {exc}"

    # ── List agents ──────────────────────────────────────────────────
    if re.search(r"^(list\s+agents?|agents?|show\s+agents?|my\s+agents?)$", text.strip(), re.I):
        try:
            r = requests.get(
                f"{DASHBOARD_URL}/api/agent-registry/agents",
                headers=DASH_HEADERS, timeout=10,
            )
            agents = r.json()
            if not agents:
                return "No agents registered yet. Use 'register agent [name]' to add one."
            lines = []
            for a in agents[:10]:
                badge = {"verified": "V", "pending": "P", "unverified": "U", "rejected": "X"}.get(a.get("verification_badge", ""), "?")
                lines.append(f"[{badge}] `{a['agent_id']}` — *{a['agent_name']}*")
            header = f"*Agent Registry* ({len(agents)} total)\n\n"
            legend = "\n\n_V=Verified P=Pending U=Unverified X=Rejected_"
            return header + "\n".join(lines) + legend
        except Exception as exc:
            return f"Failed to load agents: {exc}"

    # ── Agent status BZAG-XXXXX ──────────────────────────────────────
    m = re.search(r"\bagent\s+status\s+(BZAG-[A-Z0-9]{5})\b", text, re.I)
    if m:
        agent_id = m.group(1).upper()
        try:
            r = requests.get(
                f"{DASHBOARD_URL}/api/agent-registry/agent/{agent_id}",
                headers=DASH_HEADERS, timeout=10,
            )
            if r.status_code == 404:
                return f"Agent `{agent_id}` not found."
            a = r.json()
            status_emoji = {"verified": "CA Verified", "pending": "Pending", "unverified": "Unverified", "rejected": "Rejected"}.get(a.get("verification_status", ""), a.get("verification_status", ""))
            owner_type = "Legal Entity" if a.get("owner_type") == "legal_entity" else "Individual"
            lines = [
                f"*Agent: {a.get('agent_name', '')}*",
                f"ID: `{agent_id}`",
                f"Status: {status_emoji}",
                f"Owner: {a.get('owner_name', '')} ({owner_type})",
            ]
            if a.get("agent_description"):
                lines.append(f"Description: {a['agent_description'][:200]}")
            if a.get("verification_date"):
                lines.append(f"Verified: {a['verification_date'][:10]}")
            if a.get("ca_certificate_id"):
                lines.append(f"Cert ID: `{a['ca_certificate_id']}`")
            return "\n".join(lines)
        except Exception as exc:
            return f"Failed to get agent status: {exc}"

    # ── Support tickets ──────────────────────────────────────────────
    if re.search(r"\b(support\s+ticket|open\s+ticket|ticket\s+list|recent\s+ticket)", text, re.I):
        try:
            r = requests.get(
                f"{DASHBOARD_URL}/api/support/stats",
                headers=DASH_HEADERS, timeout=10,
            )
            stats = r.json() if r.status_code < 300 else {}
            r2 = requests.get(
                f"{DASHBOARD_URL}/api/support/tickets",
                params={"status": "open", "limit": 5},
                headers=DASH_HEADERS, timeout=10,
            )
            tickets = r2.json() if r2.status_code < 300 else []
            lines = [
                f"*Support Tickets*\n",
                f"Open: {stats.get('open', 0)} | "
                f"Auto-resolved: {stats.get('auto_resolved', 0)} | "
                f"Total: {stats.get('total', 0)}",
                f"Resolution rate: {stats.get('resolution_rate', 0)}%",
                f"Today: {stats.get('today', 0)} new\n",
            ]
            if tickets:
                lines.append("*Recent Open Tickets:*")
                for t in tickets[:5]:
                    tid = t.get("id", "?")
                    subj = (t.get("subject") or "No subject")[:40]
                    frm = (t.get("email") or "?")
                    cat = t.get("issue_category") or "undetected"
                    lines.append(f"  #{tid} [{cat}] {subj} — {frm}")
            else:
                lines.append("No open tickets.")
            return "\n".join(lines)
        except Exception as exc:
            return f"Failed to fetch support tickets: {exc}"

    if re.search(r"\bsupport\s+stats?\b", text, re.I):
        try:
            r = requests.get(
                f"{DASHBOARD_URL}/api/support/stats",
                headers=DASH_HEADERS, timeout=10,
            )
            stats = r.json() if r.status_code < 300 else {}
            common = stats.get("common_issues", [])
            lines = [
                f"*Support Statistics*\n",
                f"Total tickets: {stats.get('total', 0)}",
                f"Open: {stats.get('open', 0)}",
                f"Auto-resolved: {stats.get('auto_resolved', 0)}",
                f"Resolved: {stats.get('resolved', 0)}",
                f"Forwarded: {stats.get('forwarded', 0)}",
                f"Today: {stats.get('today', 0)}",
                f"Resolution rate: {stats.get('resolution_rate', 0)}%",
            ]
            if common:
                lines.append("\n*Common Issues:*")
                for ci in common[:5]:
                    lines.append(f"  - {ci.get('issue_category', '?')}: {ci.get('c', 0)} tickets")
            return "\n".join(lines)
        except Exception as exc:
            return f"Failed to fetch support stats: {exc}"

    # ── No action matched — let RAG handle it ─────────────────────────
    return None


async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    # Only handle private messages
    if update.effective_chat and update.effective_chat.type != "private":
        return

    user_id  = update.effective_user.id
    username = update.effective_user.username or ""
    text     = (update.message.text or "").strip()

    if not text:
        return

    if not is_owner(user_id):
        limited, reason = is_rate_limited(user_id)
        if limited:
            if reason == "long":
                await update.message.reply_text(
                    "You've reached the message limit (10 per 6 hours). "
                    "Please try again later, or contact the business owner for priority access."
                )
            else:
                await update.message.reply_text(
                    f"Please slow down — max {BURST_MAX_REQUESTS} messages per {BURST_WINDOW_SECONDS}s."
                )
            return

    # ── /help intake (takes priority over everything for users mid-intake) ────
    if await _handle_help_intake(update, user_id, text):
        return

    # ── OWNER PATH ────────────────────────────────────────────────────────────
    if is_owner(user_id):
        logger.info("OWNER MSG telegram_id=%s text=%.80s", user_id, text)

        # Try owner action first (instant — no LLM needed)
        action_result = _execute_owner_action(text, user_id)
        if action_result:
            logger.info("OWNER ACTION executed for: %.80s", text)
            if action_result == "__SHOW_OWNER_MENU__":
                await update.message.reply_text(
                    "*BizNode Owner Menu*\n\nTap any category below:",
                    parse_mode="Markdown",
                    reply_markup=_owner_menu(),
                )
                return
            try:
                await update.message.reply_text(action_result, parse_mode="Markdown")
            except Exception:
                await update.message.reply_text(action_result)
            return

        # No action matched — fall through to Tier 2 (LLM tool calling)
        thinking = await update.message.reply_text("⏳ Thinking...")
        try:
            loop = asyncio.get_event_loop()
            response = await loop.run_in_executor(None, owner_tool_answer, user_id, text, username)
            await thinking.edit_text(response)
        except Exception as exc:
            logger.error("Tool-calling failed for owner: %s", exc, exc_info=True)
            await thinking.edit_text(f"Error generating response: {exc}")
        return

    # ── CUSTOMER PATH ─────────────────────────────────────────────────────────
    if is_blocked(user_id):
        await update.message.reply_text(
            "Your access has been restricted. Contact the business owner for help."
        )
        return

    user_rec = get_bot_user(user_id)
    step     = user_rec.get("step") or STEP_NEW

    logger.info("CUSTOMER MSG telegram_id=%s step=%s text=%.80s", user_id, step, text)

    # Onboarding state machine
    if step in (STEP_NEW, ""):
        await _onboard_start(update, user_id, username)
        return

    if step == STEP_AWAITING_NAME:
        await _handle_awaiting_name(update, user_id, text)
        return

    if step == STEP_AWAITING_BUSINESS:
        await _handle_awaiting_business(update, user_id, text)
        return

    if step == STEP_AWAITING_EMAIL:
        await _handle_awaiting_email(update, user_id, text, user_rec)
        return

    if step == STEP_AWAITING_OTP:
        await _handle_awaiting_otp(update, user_id, text, user_rec)
        return

    # Verified customer — check active workflow first, then RAG
    if step == STEP_VERIFIED:
        wf_response = advance_workflow(user_id, text)
        if wf_response:
            await update.message.reply_text(wf_response)
            log_query(user_id, username, text, wf_response)
            return

        thinking = await update.message.reply_text("⏳ Thinking...")
        try:
            loop = asyncio.get_event_loop()
            response = await loop.run_in_executor(None, rag_answer, user_id, text, username)
            await thinking.edit_text(response)
        except Exception as exc:
            logger.error("RAG failed for customer: %s", exc, exc_info=True)
            await thinking.edit_text(f"Sorry, I encountered an error. Please try again.")
        return

    # Unknown step — restart
    logger.warning("Unknown step '%s' for user %s — restarting onboarding", step, user_id)
    await _onboard_start(update, user_id, username)


# ---------------------------------------------------------------------------
# Info callback handler — instant responses (no LLM)
# ---------------------------------------------------------------------------

async def _handle_info_callback(query, user_id: int, sub: str) -> None:
    """Handle info: callbacks with instant, pre-written content."""
    profile  = get_profile()
    biz_name = (profile.get("business_name") or "BizNode").strip()
    persona  = (profile.get("ai_persona_name") or "BizNode AI").strip()

    if sub == "about_biznode":
        text = (
            f"*About {persona} — AI-Powered Business Bot*\n\n"
            f"I'm the AI assistant for *{biz_name}*, available *24/7*.\n\n"
            "*Why AI + Bot?*\n"
            "- Instant responses — no waiting for business hours\n"
            "- Automated lead generation via email & groups\n"
            "- Smart price negotiation & service matching\n"
            "- Business research to grow YOUR business\n"
            "- Market making — I actively find opportunities\n"
            "- Group advertising & affiliate link publishing\n"
            "- Inter-bot communication for business automation\n\n"
            "*BizNode Platform*\n"
            "Register your bot free at https://biznode.1bz.biz\n"
            "Get DZIT credits for advanced features: https://dzit.1bz.biz\n\n"
            "_DZIT credits power blockchain-based bot-to-bot transactions "
            "and premium AI services._"
        )
    elif sub == "my_services":
        text = (
            "*Your Services*\n\n"
            "As a valued customer, here's what I can do for you:\n"
            "- Answer any question about our products/services\n"
            "- Request price quotes & negotiate on your behalf\n"
            "- Research to help grow YOUR business\n"
            "- Daily email updates with opportunities\n"
            "- Connect you with service providers & partners\n\n"
            "Just type your question or select from the menu!"
        )
    elif sub == "price_quote":
        text = (
            "*Request a Price Quote*\n\n"
            "Tell me what service or product you're interested in, "
            "and I'll prepare a personalised quote.\n\n"
            "Just type your request — e.g.:\n"
            "_\"I need pricing for web development\"_\n"
            "_\"What's the cost for monthly AI consulting?\"_"
        )
    elif sub == "biz_overview":
        text = (
            f"*{biz_name} — Business Overview*\n\n"
            "I can provide real-time metrics and growth data.\n"
            "Ask me about:\n"
            "- Customer acquisition & lead pipeline\n"
            "- Service utilisation & revenue trends\n"
            "- Market positioning & competitive analysis\n\n"
            "Type your question for detailed insights."
        )
    elif sub == "growth":
        report = get_report()
        text = (
            f"*Growth Metrics*\n\n"
            f"Total leads: *{report.get('total_leads', 0)}*\n"
            f"Verified customers: *{report.get('verified_customers', 0)}*\n"
            f"Bot users: *{report.get('total_bot_users', 0)}*\n"
            f"Queries today: *{report.get('queries_today', 0)}*\n\n"
            "Ask me for deeper analysis on any metric."
        )
    elif sub == "partnership":
        text = (
            "*Partnership Opportunities*\n\n"
            f"Interested in partnering with *{biz_name}*?\n\n"
            "We offer:\n"
            "- API integration for your services\n"
            "- Affiliate programme with commission\n"
            "- White-label bot deployment\n"
            "- Cross-promotion to our customer base\n\n"
            "Type your proposal or question!"
        )
    elif sub == "finance":
        text = (
            "*Financial Overview*\n\n"
            "For financial discussions, I can provide:\n"
            "- Revenue model & unit economics\n"
            "- Customer acquisition cost data\n"
            "- Growth projections & market size\n\n"
            "Ask me for specific financial data."
        )
    elif sub == "services":
        # Show available services with buy buttons
        data = _dash_get("/api/services")
        if isinstance(data, list) and data:
            lines = ["*Our Services:*\n"]
            keyboard = []
            for svc in data:
                if not svc.get("is_active"):
                    continue
                name = svc.get("name", "Service")
                price = svc.get("price", 0)
                currency = svc.get("currency", "USD")
                desc = svc.get("description", "")
                lines.append(f"*{name}* — {currency} {price:.2f}")
                if desc:
                    lines.append(f"  _{desc}_\n")
                keyboard.append([InlineKeyboardButton(
                    f"Buy: {name}", callback_data=f"buy:{svc['id']}"
                )])
            await query.message.reply_text(
                "\n".join(lines), parse_mode="Markdown",
                reply_markup=InlineKeyboardMarkup(keyboard) if keyboard else None,
            )
        else:
            await query.message.reply_text("No services listed yet. Contact us for details!")
        return
    elif sub == "buy_license":
        # Show license purchase options
        keyboard = [
            [InlineKeyboardButton("Buy Basic ($200 USD / 200 BZeUSD)", callback_data="license:stripe:basic")],
            [InlineKeyboardButton("Buy Basic (Crypto)", callback_data="license:crypto:basic")],
            [InlineKeyboardButton("Buy Pro ($499 USD / 499 BZeUSD)", callback_data="license:stripe:pro")],
            [InlineKeyboardButton("Buy Pro (Crypto)", callback_data="license:crypto:pro")],
        ]
        await query.message.reply_text(
            "*BizNode License*\n\n"
            "*Basic* — $200 USD or 200 BZeUSD\n"
            "  1 bot, core features, software key\n\n"
            "*Pro* — $499 USD or 499 BZeUSD\n"
            "  Up to 5 bots, all features, USB key\n\n"
            "Choose your license:",
            parse_mode="Markdown",
            reply_markup=InlineKeyboardMarkup(keyboard),
        )
        return
    elif sub == "affiliates":
        data = _dash_get("/api/affiliate-links", active_only=True)
        if isinstance(data, list) and data:
            lines = ["*Partner Links & Resources:*\n"]
            keyboard = []
            for link in data[:8]:
                title = link.get("title", "Link")
                url = link.get("url", "")
                desc = link.get("description", "")
                commission = link.get("commission", "")
                line = f"- *{title}*"
                if desc:
                    line += f" — {desc}"
                if commission:
                    line += f" _{commission}_"
                lines.append(line)
                keyboard.append([InlineKeyboardButton(title, url=url)])
            await query.message.reply_text(
                "\n".join(lines), parse_mode="Markdown",
                reply_markup=InlineKeyboardMarkup(keyboard) if keyboard else None,
                disable_web_page_preview=True,
            )
        else:
            await query.message.reply_text("No partner links available yet.")
        return
    elif sub == "change_role":
        user_rec = get_bot_user(user_id)
        current = user_rec.get("person_type") or PERSON_VISITOR
        label = PERSON_TYPE_LABELS.get(current, current)
        await query.message.reply_text(
            f"Current role: *{label}*\n\nSelect your new role:",
            parse_mode="Markdown",
            reply_markup=_role_selection_menu(),
        )
        return
    else:
        text = "Information not available. Type your question and I'll help!"

    await query.message.reply_text(text, parse_mode="Markdown",
                                    disable_web_page_preview=True)


# ---------------------------------------------------------------------------
# Transaction / Payment / Buy callback handlers
# ---------------------------------------------------------------------------

async def _handle_txn_callback(query, user_id: int, sub: str) -> None:
    """Handle txn: callbacks for transaction creation and history."""
    if sub == "history":
        data = _dash_get(f"/api/transactions", telegram_id=str(user_id), limit=10)
        if not isinstance(data, list) or not data:
            await query.message.reply_text("No transactions yet.")
            return
        lines = ["*Your Transactions:*\n"]
        for t in data[:10]:
            ref = t.get("reference_no", "?")
            title = t.get("title", "?")
            status = t.get("status", "?")
            amount = t.get("amount", 0)
            currency = t.get("currency", "USD")
            line = f"`{ref}` — {title}\n  {currency} {amount:.2f} [{status}]"
            lines.append(line)
        await query.message.reply_text("\n\n".join(lines), parse_mode="Markdown")
        return

    # Create a transaction of the selected type
    from services.transaction_service import TRANSACTION_LABELS
    label = TRANSACTION_LABELS.get(sub, sub)
    user_rec = get_bot_user(user_id)
    person_type = user_rec.get("person_type") or PERSON_VISITOR
    name = user_rec.get("name") or "Customer"

    result = _dash_post("/api/transactions", {
        "telegram_id": str(user_id),
        "person_type": person_type,
        "transaction_type": sub,
        "title": label,
        "description": f"Initiated by {name} via Telegram",
    })

    if result.get("reference_no"):
        ref = result["reference_no"]
        if sub == "price_negotiation":
            await query.message.reply_text(
                f"*{label}* started (Ref: `{ref}`)\n\n"
                "Use /negotiate ServiceName YourPrice to submit your offer.",
                parse_mode="Markdown",
            )
        elif sub in ("stripe_payment", "crypto_payment"):
            await query.message.reply_text(
                f"*{label}* started (Ref: `{ref}`)\n\n"
                "Use /pay to choose your payment method.",
                parse_mode="Markdown",
            )
        else:
            await query.message.reply_text(
                f"*{label}* initiated! (Ref: `{ref}`)\n\n"
                "Type your question or request, and I'll process it.",
                parse_mode="Markdown",
            )
    else:
        await query.message.reply_text("Could not create transaction. Try again.")


async def _handle_pay_callback(query, user_id: int, sub: str) -> None:
    """Handle pay: callbacks for payment method selection."""
    if sub == "stripe":
        stripe_status = _dash_get("/api/stripe/status")
        if stripe_status.get("configured"):
            await query.message.reply_text(
                "*Stripe Card Payment*\n\n"
                "Use /services to browse and buy with card.\n\n"
                "Or type: `/pay_stripe ServiceName Amount`",
                parse_mode="Markdown",
            )
        else:
            await query.message.reply_text(
                "Stripe payments are not yet configured.\n"
                "Please contact the business owner or use crypto payment."
            )

    elif sub == "crypto":
        info = _dash_get(f"/api/crypto/wallet/{user_id}")
        if info.get("registered"):
            verified = "Yes" if info.get("verified") else "No"
            await query.message.reply_text(
                f"*Crypto Payment (Polygon MATIC)*\n\n"
                f"Your wallet: `{info.get('wallet', '')}`\n"
                f"Verified: *{verified}*\n\n"
                "Use /verify\\_wallet after sending payment.",
                parse_mode="Markdown",
            )
        else:
            await query.message.reply_text(
                "*Crypto Payment (Polygon MATIC)*\n\n"
                "First, register your wallet:\n"
                "`/wallet 0xYourPolygonAddress`\n\n"
                "Then send the payment and use /verify\\_wallet.",
                parse_mode="Markdown",
            )

    elif sub == "history":
        data = _dash_get(f"/api/transactions", telegram_id=str(user_id), limit=10)
        payments = [t for t in (data if isinstance(data, list) else [])
                    if t.get("transaction_type") in ("stripe_payment", "crypto_payment")]
        if not payments:
            await query.message.reply_text("No payment transactions yet.")
            return
        lines = ["*Payment History:*\n"]
        for t in payments[:10]:
            ref = t.get("reference_no", "?")
            title = t.get("title", "?")
            status = t.get("status", "?")
            amount = t.get("amount", 0)
            currency = t.get("currency", "USD")
            lines.append(f"`{ref}` — {title}\n  {currency} {amount:.2f} [{status}]")
        await query.message.reply_text("\n\n".join(lines), parse_mode="Markdown")


async def _handle_buy_callback(query, user_id: int, service_id: str) -> None:
    """Handle buy:<service_id> callbacks — show payment options for a service."""
    data = _dash_get("/api/services")
    service = None
    for s in (data if isinstance(data, list) else []):
        if str(s.get("id")) == service_id:
            service = s
            break

    if not service:
        await query.message.reply_text("Service not found.")
        return

    name = service.get("name", "Service")
    price = service.get("price", 0)
    currency = service.get("currency", "USD")

    keyboard = [
        [InlineKeyboardButton(
            f"Pay {currency} {price:.2f} with Card",
            callback_data=f"pay:stripe"
        )],
        [InlineKeyboardButton(
            f"Pay with Crypto (MATIC)",
            callback_data=f"pay:crypto"
        )],
    ]

    await query.message.reply_text(
        f"*{name}*\n"
        f"Price: *{currency} {price:.2f}*\n\n"
        "Choose payment method:",
        parse_mode="Markdown",
        reply_markup=InlineKeyboardMarkup(keyboard),
    )


# ---------------------------------------------------------------------------
# Group ad sender (called from bot context with access to bot object)
# ---------------------------------------------------------------------------

async def _send_pending_group_ads(context: ContextTypes.DEFAULT_TYPE) -> None:
    """Send any pending group ads. Called on bot startup or scheduled."""
    if not TELEGRAM_GROUP_ID:
        return
    try:
        from services.group_ad_service import get_pending_ads, mark_ad_sent
        ads = get_pending_ads()
        for ad in ads[:3]:  # Max 3 ads per cycle
            message = ad.get("message", "")
            ad_type = ad.get("ad_type", "promotion")
            if ad_type == "opportunity":
                prefix = "Business Opportunity\n\n"
            elif ad_type == "affiliate":
                prefix = "Recommended\n\n"
            elif ad_type == "alert":
                prefix = "Alert\n\n"
            else:
                prefix = "Announcement\n\n"
            try:
                await context.bot.send_message(
                    chat_id=TELEGRAM_GROUP_ID,
                    text=prefix + message,
                )
                mark_ad_sent(ad["id"])
                logger.info("Group ad sent: id=%s title=%s", ad["id"], ad.get("title"))
            except Exception as exc:
                logger.error("Failed to send group ad %s: %s", ad["id"], exc)
    except Exception as exc:
        logger.error("Group ad check failed: %s", exc)


async def _send_daily_business_update(context: ContextTypes.DEFAULT_TYPE) -> None:
    """Post a daily business update to the Telegram group with website links.
    Runs once per day. Uses LLM to compose a fresh, engaging update."""
    if not TELEGRAM_GROUP_ID:
        return
    try:
        # Gather business stats for the update
        stats = _dash_get("/api/status") or {}
        total_users = stats.get("total_users", 0)
        total_leads = stats.get("total_leads", 0)

        # Compose a proactive daily update with website links
        from datetime import date
        today = date.today().strftime("%B %d, %Y")
        site_url = "https://biznode.1bz.biz"

        update_text = (
            f"*BizNode Daily Update — {today}*\n\n"
            f"Your AI-powered business automation platform is active and growing.\n\n"
            f"*What's New:*\n"
            f"• AI Smart Bot — automates customer engagement 24/7\n"
            f"• BizChannel Marketplace — connect with businesses\n"
            f"• Payment Agent — Stripe & crypto payments built-in\n\n"
            f"*Quick Links:*\n"
            f"• [Get Started]({site_url}) — Download BizNode\n"
            f"• [Features]({site_url}/features.php) — See all features\n"
            f"• [Tutorial]({site_url}/tutorial.php) — Step-by-step guide\n"
            f"• [Update Guide]({site_url}/update-guide.php) — Latest updates\n\n"
            f"_Automate your business. Start today._\n"
            f"@biznode\\_bot — Talk to our AI assistant"
        )

        await context.bot.send_message(
            chat_id=TELEGRAM_GROUP_ID,
            text=update_text,
            parse_mode="Markdown",
            disable_web_page_preview=True,
        )
        logger.info("Daily business update posted to group")
    except Exception as exc:
        logger.error("Daily business update failed: %s", exc)


# ---------------------------------------------------------------------------
# /support — AI-Powered Remote Support via MCP
# ---------------------------------------------------------------------------

async def cmd_support(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """AI-powered support: diagnose and fix customer nodes via MCP. /support"""
    user_id = update.effective_user.id
    text = (update.message.text or "").strip()
    parts = text.split(maxsplit=1)
    issue_desc = parts[1] if len(parts) > 1 else ""

    # Look up customer's registered node
    node_info = _find_customer_node(user_id)

    if not node_info:
        await update.message.reply_text(
            "*Remote Support*\n\n"
            "No registered node found for your account.\n\n"
            "If you have a local node (BizMax/BizNode/1BZNode), make sure:\n"
            "1. Your node is running\n"
            "2. Port is open (7877 basic, 7977 pro)\n"
            "3. Node registered with @biznode_bot via /setup\n\n"
            "For API-only tiers, describe your issue and our AI will help.",
            parse_mode="Markdown",
        )
        if issue_desc:
            # Handle as general support question via LLM
            await _handle_support_question(update, issue_desc)
        return

    target = node_info["endpoint"]
    await update.message.reply_text(
        f"*Checking your node...*\nTarget: `{target}`",
        parse_mode="Markdown",
    )

    # Run MCP diagnostics
    diag = await _run_node_diagnostics(target)

    # AI analyzes results and suggests fixes
    analysis = _analyze_diagnostics(diag, issue_desc)

    # Attempt auto-fix if possible
    fix_result = ""
    if analysis.get("auto_fix"):
        fix_result = await _attempt_auto_fix(target, analysis["auto_fix"])

    # Build response
    response = f"*Node Diagnostics*\n`{target}`\n\n"

    # Service status
    services = diag.get("services", {})
    for svc, status in services.items():
        icon = "\u2705" if status == "up" else "\u274c"
        response += f"{icon} {svc}: {status}\n"

    response += f"\n*Analysis:*\n{analysis.get('message', 'No issues detected.')}\n"

    if fix_result:
        response += f"\n*Auto-Fix:*\n{fix_result}\n"

    if analysis.get("suggestions"):
        response += "\n*Suggestions:*\n"
        for s in analysis["suggestions"]:
            response += f"\u2022 {s}\n"

    await update.message.reply_text(response, parse_mode="Markdown")


def _find_customer_node(user_id: int) -> dict:
    """Find a customer's registered node by their Telegram user ID."""
    try:
        _project = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
        sys.path.insert(0, _project)
        from memory.database import get_connection
        conn = get_connection()
        cursor = conn.cursor()
        # Check hosted_agents by telegram_id
        cursor.execute(
            "SELECT agent_id, bot_username, tier FROM hosted_agents WHERE telegram_id = ? AND status = 'active' LIMIT 1",
            (str(user_id),),
        )
        row = cursor.fetchone()
        if row:
            agent = dict(row)
            tier = agent.get("tier", "basic")
            # API-only tiers have no remote node
            if tier in ("basic", "standard", "pro", "bizplus"):
                conn.close()
                return None
            # Node tiers — look up endpoint from agent_registry
            cursor.execute(
                "SELECT agent_endpoint FROM agent_registry WHERE telegram_bot_username = ? LIMIT 1",
                (agent.get("bot_username", ""),),
            )
            ep_row = cursor.fetchone()
            conn.close()
            if ep_row and dict(ep_row).get("agent_endpoint"):
                return {"endpoint": dict(ep_row)["agent_endpoint"], "agent_id": agent["agent_id"], "tier": tier}

        # Also check client registrations
        cursor.execute(
            "SELECT * FROM businesses WHERE owner_telegram_id = ? LIMIT 1",
            (str(user_id),),
        )
        biz_row = cursor.fetchone()
        conn.close()
        if biz_row:
            biz = dict(biz_row)
            node_id = biz.get("node_id", "")
            # Try to find endpoint from trusted_nodes
            conn2 = get_connection()
            cursor2 = conn2.cursor()
            cursor2.execute("SELECT endpoint FROM trusted_nodes WHERE node_id = ? LIMIT 1", (node_id,))
            tn_row = cursor2.fetchone()
            conn2.close()
            if tn_row and dict(tn_row).get("endpoint"):
                return {"endpoint": dict(tn_row)["endpoint"], "node_id": node_id}

    except Exception:
        pass
    return None


async def _run_node_diagnostics(target: str) -> dict:
    """Run MCP health check on a remote node."""
    try:
        _project = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
        sys.path.insert(0, _project)
        from tools.admin.remote_control import send_command
        result = send_command(target, "health_check")
        if result.get("success"):
            return result["response"]["result"]
        return {"error": result.get("error", "MCP command failed")}
    except Exception as exc:
        return {"error": str(exc)}


def _analyze_diagnostics(diag: dict, issue_desc: str = "") -> dict:
    """AI analysis of diagnostic results — determines fixes and suggestions."""
    services = diag.get("services", {})
    auto_fix = []
    suggestions = []
    problems = []

    if diag.get("error"):
        return {
            "message": f"Could not reach your node: {diag['error']}",
            "suggestions": [
                "Check that your node is running",
                "Make sure port is open and accessible",
                "Try restarting your dashboard: python launcher.py",
            ],
        }

    # Check each service
    if services.get("dashboard") == "down":
        problems.append("Dashboard is down")
        auto_fix.append({"command": "restart_service", "params": {"service": "dashboard"}})
        suggestions.append("Dashboard will be restarted automatically")

    if services.get("postgresql") == "down":
        problems.append("PostgreSQL database is down")
        suggestions.append("Start PostgreSQL: net start postgresql-x64-18 (or your version)")
        suggestions.append("Check pg_hba.conf for connection settings")

    if services.get("ollama") == "down":
        problems.append("Ollama AI model server is down")
        suggestions.append("Start Ollama: ollama serve")
        suggestions.append("Or install from https://ollama.com")

    if services.get("qdrant") == "down":
        problems.append("Qdrant vector DB is not running")
        suggestions.append("Qdrant may not be installed — this is optional for basic nodes")
        suggestions.append("Install with Docker: docker run -p 6333:6333 qdrant/qdrant")

    all_up = all(v == "up" for v in services.values())

    if all_up and not issue_desc:
        return {"message": "All services are running. Your node is healthy!", "suggestions": []}

    if all_up and issue_desc:
        return {
            "message": f"All services are running. Your issue: \"{issue_desc}\" may need manual investigation.",
            "suggestions": [
                "Try restarting the specific service that's causing issues",
                "Check logs with: /support logs",
                "Contact biznode@1bz.biz for advanced support",
            ],
        }

    msg = "Issues found:\n" + "\n".join(f"- {p}" for p in problems)
    return {"message": msg, "auto_fix": auto_fix, "suggestions": suggestions}


async def _attempt_auto_fix(target: str, fixes: list) -> str:
    """Attempt automatic fixes via MCP."""
    results = []
    try:
        _project = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
        sys.path.insert(0, _project)
        from tools.admin.remote_control import send_command
        for fix in fixes:
            r = send_command(target, fix["command"], fix.get("params", {}))
            if r.get("success"):
                results.append(f"\u2705 {fix['command']}: Success")
            else:
                results.append(f"\u274c {fix['command']}: {r.get('error', 'Failed')}")
    except Exception as exc:
        results.append(f"\u274c Auto-fix error: {exc}")
    return "\n".join(results) if results else ""


async def _handle_support_question(update: Update, question: str) -> None:
    """Handle a general support question via KB search + LLM."""
    try:
        response = await _ask_rag(question)
        if response:
            await update.message.reply_text(response, parse_mode="Markdown")
        else:
            await update.message.reply_text(
                "I couldn't find a specific answer. Please email biznode@1bz.biz or "
                "describe your issue in more detail.",
            )
    except Exception:
        await update.message.reply_text(
            "Support system is temporarily unavailable. Please email biznode@1bz.biz",
        )


# ---------------------------------------------------------------------------
# /setup — Node Setup Wizard
# ---------------------------------------------------------------------------

async def cmd_setup(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Start BizNode node setup wizard. /setup"""
    user_id = update.effective_user.id

    _setup_wizard[user_id] = {"step": "tier"}

    keyboard = [
        [InlineKeyboardButton("Basic — $20 (API only)", callback_data="setup:tier:basic")],
        [InlineKeyboardButton("Standard — $50 (5 bots)", callback_data="setup:tier:standard")],
        [InlineKeyboardButton("Pro — $100 (escrow + earn)", callback_data="setup:tier:pro")],
        [InlineKeyboardButton("BizPlus — $150 (5 workflows)", callback_data="setup:tier:bizplus")],
        [InlineKeyboardButton("BizMax — $200 (local node)", callback_data="setup:tier:bizmax")],
        [InlineKeyboardButton("BizNode — $499 (USB + full)", callback_data="setup:tier:biznode")],
        [InlineKeyboardButton("1BZNode — $1500 (1500 DZIT)", callback_data="setup:tier:onezbnode")],
    ]

    await update.message.reply_text(
        "*BizNode Node Setup Wizard*\n\n"
        "I'll help you set up your AI agent.\n\n"
        "*Step 1 — Choose your tier:*\n\n"
        "*Basic* $20 — 1 bot, API-hosted, 10 DZIT\n"
        "*Standard* $50 — 5 bots, API-hosted\n"
        "*Pro* $100 — Escrow payments, earn DZIT\n"
        "*BizPlus* $150 — 5 published workflows\n"
        "*BizMax* $200 — Local node software\n"
        "*BizNode* $499 — USB key, full platform\n"
        "*1BZNode* $1500 — 1500 DZIT bonus\n",
        parse_mode="Markdown",
        reply_markup=InlineKeyboardMarkup(keyboard),
    )


async def _handle_setup_callback(query, user_id: int, sub: str) -> None:
    state = _setup_wizard.get(user_id)
    if not state:
        await query.message.reply_text("No active setup. Start with /setup")
        return

    parts = sub.split(":", 1)
    action = parts[0]
    value = parts[1] if len(parts) > 1 else ""

    # Step 1: Tier selection
    if action == "tier":
        state["tier"] = value
        # API-only tiers skip infrastructure steps
        api_only_tiers = ("basic", "standard", "pro", "bizplus")
        if value in api_only_tiers:
            state["db_type"] = "hosted"
            state["llm_mode"] = "shared"
            state["qdrant_mode"] = "shared"
            state["step"] = "industry"
            rows = []
            for i in range(0, len(SETUP_INDUSTRIES), 2):
                row = [InlineKeyboardButton(
                    SETUP_INDUSTRIES[i][1], callback_data=f"setup:ind:{SETUP_INDUSTRIES[i][0]}"
                )]
                if i + 1 < len(SETUP_INDUSTRIES):
                    row.append(InlineKeyboardButton(
                        SETUP_INDUSTRIES[i + 1][1], callback_data=f"setup:ind:{SETUP_INDUSTRIES[i + 1][0]}"
                    ))
                rows.append(row)
            tier_prices = {"basic": "$20", "standard": "$50", "pro": "$100", "bizplus": "$150"}
            await query.message.reply_text(
                f"*{value.upper()}* tier ({tier_prices.get(value, '')}) — API-hosted, no install needed\n\n"
                "*Step 2 — Choose your industry:*",
                parse_mode="Markdown",
                reply_markup=InlineKeyboardMarkup(rows),
            )
            return
        # Node-install tiers continue with infrastructure steps
        state["step"] = "db"
        keyboard = [
            [InlineKeyboardButton("SQLite (simplest)", callback_data="setup:db:sqlite")],
            [InlineKeyboardButton("MySQL", callback_data="setup:db:mysql")],
            [InlineKeyboardButton("PostgreSQL", callback_data="setup:db:postgresql")],
        ]
        tier_prices = {"bizmax": "$200", "biznode": "$499", "onezbnode": "$1500"}
        await query.message.reply_text(
            f"*{value.upper()}* tier ({tier_prices.get(value, '')}) — Local node install\n\n"
            f"*Step 2 — Database:*\n\n"
            f"*SQLite* — No server needed, single file\n"
            f"*MySQL* — For medium workloads\n"
            f"*PostgreSQL* — For heavy workloads\n",
            parse_mode="Markdown",
            reply_markup=InlineKeyboardMarkup(keyboard),
        )
        return

    # Step 2: Database selection
    if action == "db":
        state["db_type"] = value
        state["step"] = "llm"
        keyboard = [
            [InlineKeyboardButton("Own Ollama (local, full control)", callback_data="setup:llm:own")],
            [InlineKeyboardButton("Shared LLM (read-only, no install)", callback_data="setup:llm:shared")],
        ]
        await query.message.reply_text(
            f"*Step 3/6 — LLM (AI Model):*\n\n"
            f"*Own* — Install Ollama locally, full model control\n"
            f"*Shared* — Use our LLM gateway (chat & generate only)\n",
            parse_mode="Markdown",
            reply_markup=InlineKeyboardMarkup(keyboard),
        )
        return

    # Step 3: LLM mode
    if action == "llm":
        state["llm_mode"] = value
        state["step"] = "qdrant"
        keyboard = [
            [InlineKeyboardButton("Own Qdrant (local, full CRUD)", callback_data="setup:qd:own")],
            [InlineKeyboardButton("Shared Qdrant (search only)", callback_data="setup:qd:shared")],
        ]
        await query.message.reply_text(
            f"*Step 4/6 — Knowledge Base (Qdrant):*\n\n"
            f"*Own* — Run Qdrant locally, full create/search/delete\n"
            f"*Shared* — Search our shared knowledge base (read-only)\n",
            parse_mode="Markdown",
            reply_markup=InlineKeyboardMarkup(keyboard),
        )
        return

    # Step 4: Qdrant mode
    if action == "qd":
        state["qdrant_mode"] = value
        state["step"] = "industry"
        rows = []
        for i in range(0, len(SETUP_INDUSTRIES), 2):
            row = [InlineKeyboardButton(
                SETUP_INDUSTRIES[i][1], callback_data=f"setup:ind:{SETUP_INDUSTRIES[i][0]}"
            )]
            if i + 1 < len(SETUP_INDUSTRIES):
                row.append(InlineKeyboardButton(
                    SETUP_INDUSTRIES[i + 1][1], callback_data=f"setup:ind:{SETUP_INDUSTRIES[i + 1][0]}"
                ))
            rows.append(row)
        await query.message.reply_text(
            "*Step 5/6 — Industry:*\n\n"
            "Select your industry for customized features:",
            parse_mode="Markdown",
            reply_markup=InlineKeyboardMarkup(rows),
        )
        return

    # Step 5: Industry selection
    if action == "ind":
        state["industry"] = value
        state["step"] = "confirm"
        industry_label = dict(SETUP_INDUSTRIES).get(value, value)

        tier = state["tier"]
        db = state["db_type"]
        llm = state["llm_mode"]
        qd = state["qdrant_mode"]
        tier_info = {
            "basic": (1, "$20"), "standard": (5, "$50"), "pro": (5, "$100"),
            "bizplus": (5, "$150"), "bizmax": (1, "$200"), "biznode": (5, "$499"),
            "onezbnode": (5, "$1500"),
        }
        max_bots, price = tier_info.get(tier, (1, "$20"))

        keyboard = [
            [InlineKeyboardButton("Confirm & Request Setup", callback_data="setup:ok:go")],
            [InlineKeyboardButton("Start Over", callback_data="setup:ok:restart")],
            [InlineKeyboardButton("Cancel", callback_data="setup:ok:cancel")],
        ]

        await query.message.reply_text(
            f"*Step 6/6 — Review Your Configuration:*\n\n"
            f"Tier: *{tier.upper()}* ({price})\n"
            f"Database: *{db.upper()}*\n"
            f"LLM: *{llm.upper()}*"
            f"{' (shared gateway, read-only)' if llm == 'shared' else ' (local Ollama)'}\n"
            f"Knowledge Base: *{qd.upper()}*"
            f"{' (shared, search-only)' if qd == 'shared' else ' (local Qdrant)'}\n"
            f"Industry: *{industry_label}*\n"
            f"Max Bots: *{max_bots}*\n"
            f"USB Required: *{'Yes' if tier == 'pro' else 'No'}*\n",
            parse_mode="Markdown",
            reply_markup=InlineKeyboardMarkup(keyboard),
        )
        return

    # Step 6: Confirmation
    if action == "ok":
        if value == "cancel":
            _setup_wizard.pop(user_id, None)
            await query.message.reply_text("Setup cancelled.")
            return

        if value == "restart":
            _setup_wizard.pop(user_id, None)
            keyboard = [
                [InlineKeyboardButton("Basic — $20 (API only)", callback_data="setup:tier:basic")],
                [InlineKeyboardButton("Standard — $50 (5 bots)", callback_data="setup:tier:standard")],
                [InlineKeyboardButton("Pro — $100 (escrow + earn)", callback_data="setup:tier:pro")],
                [InlineKeyboardButton("BizPlus — $150 (5 workflows)", callback_data="setup:tier:bizplus")],
                [InlineKeyboardButton("BizMax — $200 (local node)", callback_data="setup:tier:bizmax")],
                [InlineKeyboardButton("BizNode — $499 (USB + full)", callback_data="setup:tier:biznode")],
                [InlineKeyboardButton("1BZNode — $1500 (1500 DZIT)", callback_data="setup:tier:onezbnode")],
            ]
            _setup_wizard[user_id] = {"step": "tier"}
            await query.message.reply_text(
                "*BizNode Node Setup Wizard*\n\n"
                "*Step 1/6 — Choose your tier:*",
                parse_mode="Markdown",
                reply_markup=InlineKeyboardMarkup(keyboard),
            )
            return

        if value == "go":
            user_rec = get_bot_user(user_id)
            username = query.from_user.username or ""
            full_name = query.from_user.full_name or ""
            email = (user_rec or {}).get("email", "")

            if not email:
                _setup_wizard.pop(user_id, None)
                await query.message.reply_text(
                    "*Email Required*\n\n"
                    "You need to verify your email before requesting a setup.\n"
                    "The installer files will be sent to your email.\n\n"
                    "Type your email address to register, then run /setup again.",
                    parse_mode="Markdown",
                )
                return

            tier = state.get("tier", "basic")
            db_type = state.get("db_type", "sqlite")
            llm_mode = state.get("llm_mode", "own")
            qdrant_mode = state.get("qdrant_mode", "own")
            industry = state.get("industry", "general")
            industry_label = dict(SETUP_INDUSTRIES).get(industry, industry)
            tier_info = {
                "basic": (1, "$20"), "standard": (5, "$50"), "pro": (5, "$100"),
                "bizplus": (5, "$150"), "bizmax": (1, "$200"), "biznode": (5, "$499"),
                "onezbnode": (5, "$1500"),
            }
            max_bots, price = tier_info.get(tier, (1, "$20"))
            node_id = username or f"user-{user_id}"

            _setup_wizard.pop(user_id, None)

            await query.message.reply_text(
                "*Building your custom installer...*\n"
                "This may take a moment.",
                parse_mode="Markdown",
            )

            # Build installer package in background
            build_cmd = [
                sys.executable, os.path.join(
                    os.path.dirname(os.path.dirname(os.path.dirname(
                        os.path.abspath(__file__)))),
                    "scripts", "deploy_node.py",
                ),
                "--tier", tier,
                "--db-type", db_type,
                "--llm-mode", llm_mode,
                "--qdrant-mode", qdrant_mode,
                "--industry", industry,
                "--customer-name", full_name,
                "--customer-email", email,
                "--node-id", node_id,
                "--generate-installer",
                "--upload-only",
            ]

            _NO_WIN = 0x08000000 if sys.platform == "win32" else 0
            try:
                result = subprocess.run(
                    build_cmd, capture_output=True, text=True,
                    timeout=120, creationflags=_NO_WIN,
                )
                build_ok = result.returncode == 0
            except Exception as exc:
                logger.error("Setup build failed: %s", exc)
                build_ok = False

            if build_ok:
                # Find the generated installer and zip
                scripts_dir = os.path.dirname(os.path.dirname(
                    os.path.dirname(os.path.abspath(__file__))))
                dist_dir = os.path.join(scripts_dir, "dist")
                installer_path = os.path.join(dist_dir, f"install_{node_id}.py")
                setup_path = os.path.join(scripts_dir, "scripts", "setup_node.py")

                import glob as _glob
                zips = sorted(
                    _glob.glob(os.path.join(dist_dir, f"biznode-{tier}-*.zip")),
                    key=os.path.getmtime, reverse=True,
                )
                zip_path = zips[0] if zips else None

                # Email the installer files
                email_sent = False
                if os.path.isfile(installer_path) and zip_path and os.path.isfile(zip_path):
                    try:
                        from services.email_service import EmailService
                        import smtplib
                        from email.mime.multipart import MIMEMultipart
                        from email.mime.text import MIMEText
                        from email.mime.application import MIMEApplication

                        svc = EmailService()
                        msg = MIMEMultipart()
                        msg["From"] = svc.from_email
                        msg["To"] = email
                        msg["Subject"] = f"BizNode {tier.upper()} Node — Your Custom Installer"

                        body_text = (
                            f"Hi {full_name},\n\n"
                            f"Your custom BizNode {tier.upper()} node installer is ready!\n\n"
                            f"Configuration:\n"
                            f"  Tier: {tier.upper()} ({price})\n"
                            f"  Database: {db_type.upper()}\n"
                            f"  LLM: {llm_mode} {'(shared gateway)' if llm_mode == 'shared' else '(local Ollama)'}\n"
                            f"  Knowledge Base: {qdrant_mode} {'(shared, read-only)' if qdrant_mode == 'shared' else '(local Qdrant)'}\n"
                            f"  Industry: {industry_label}\n"
                            f"  Max Bots: {max_bots}\n\n"
                            f"INSTALLATION INSTRUCTIONS:\n"
                            f"==========================\n\n"
                            f"1. Save BOTH attached files to the SAME folder (e.g. D:\\)\n"
                            f"2. Also save setup_node.py to the same folder\n"
                            f"3. Open Command Prompt or PowerShell\n"
                            f"4. Run: python D:\\install_{node_id}.py\n"
                            f"5. The installer will:\n"
                            f"   - Extract files to C:\\{tier}\n"
                            f"   - Install {'MySQL' if db_type == 'mysql' else 'PostgreSQL' if db_type == 'postgresql' else 'SQLite (no install needed)'}\n"
                            f"   - Configure your .env file\n"
                            f"   {'- Install Ollama (AI model server)' if llm_mode == 'own' else '- Connect to shared LLM gateway'}\n"
                            f"   {'- Install Qdrant (knowledge base)' if qdrant_mode == 'own' else '- Connect to shared knowledge base'}\n"
                            f"   - Register with BizNode network\n"
                            f"   - Create auto-start scheduled task\n\n"
                            f"6. After installation, open: http://localhost:{'7877' if tier == 'basic' else '7977'}\n\n"
                            f"Need help? Reply to this email or message @biznode_bot on Telegram.\n\n"
                            f"— BizNode Team\n"
                        )
                        msg.attach(MIMEText(body_text, "plain"))

                        with open(installer_path, "rb") as f:
                            att = MIMEApplication(f.read(), _subtype="octet-stream")
                            att.add_header("Content-Disposition", "attachment",
                                           filename=f"install_{node_id}.py")
                            msg.attach(att)

                        with open(setup_path, "rb") as f:
                            att2 = MIMEApplication(f.read(), _subtype="octet-stream")
                            att2.add_header("Content-Disposition", "attachment",
                                            filename="setup_node.py")
                            msg.attach(att2)

                        # Note: ZIP too large for email (32MB), provide download link
                        zip_size_mb = os.path.getsize(zip_path) / (1024 * 1024)
                        if zip_size_mb < 20:
                            with open(zip_path, "rb") as f:
                                att3 = MIMEApplication(f.read(), _subtype="zip")
                                att3.add_header("Content-Disposition", "attachment",
                                                filename=os.path.basename(zip_path))
                                msg.attach(att3)

                        server = smtplib.SMTP(svc.smtp_host, svc.smtp_port)
                        server.starttls()
                        server.login(svc.smtp_user, svc.smtp_password)
                        server.sendmail(svc.from_email, [email], msg.as_string())
                        server.quit()
                        email_sent = True
                    except Exception as exc:
                        logger.error("Failed to email setup files: %s", exc)

                # Send instructions in bot
                download_note = ""
                if zip_path and os.path.getsize(zip_path) / (1024 * 1024) >= 20:
                    download_note = (
                        "\n*Note:* The package ZIP (32MB) is too large for email.\n"
                        "Download it from the link we'll send separately, "
                        "or contact us for a USB drive.\n"
                    )

                await query.message.reply_text(
                    f"*Your BizNode {tier.upper()} Installer is Ready!*\n\n"
                    f"{'Sent to: ' + _mask_email(email) if email_sent else 'Email delivery pending — contact support.'}\n"
                    f"{download_note}\n"
                    f"*Installation Steps:*\n"
                    f"1. Save all files to one folder (e.g. D:\\\\)\n"
                    f"2. Open Command Prompt\n"
                    f"3. Run: `python D:\\\\install_{node_id}.py`\n"
                    f"4. Wait for setup to complete\n"
                    f"5. Open: http://localhost:{'7877' if tier == 'basic' else '7977'}\n\n"
                    f"*Your Config:*\n"
                    f"Tier: {tier.upper()} | DB: {db_type} | LLM: {llm_mode} | "
                    f"KB: {qdrant_mode} | Industry: {industry_label}\n\n"
                    f"Questions? Just message me here!",
                    parse_mode="Markdown",
                )

            else:
                await query.message.reply_text(
                    "*Setup build is being processed.*\n\n"
                    "Our team has been notified and will send your "
                    "installer files shortly.\n\n"
                    "You'll receive them at: " + _mask_email(email),
                    parse_mode="Markdown",
                )

            # Notify owner
            if OWNER_ID:
                owner_msg = (
                    f"*New Setup Request*\n\n"
                    f"From: *{full_name}* (@{username})\n"
                    f"Chat ID: `{user_id}`\n"
                    f"Email: {_mask_email(email)}\n"
                    f"Build: {'OK' if build_ok else 'FAILED'}\n"
                    f"Email sent: {'Yes' if email_sent else 'No'}\n\n"
                    f"*Configuration:*\n"
                    f"Tier: *{tier.upper()}* ({price})\n"
                    f"Database: *{db_type.upper()}*\n"
                    f"LLM: *{llm_mode}*\n"
                    f"Qdrant: *{qdrant_mode}*\n"
                    f"Industry: *{industry_label}*\n"
                    f"Max Bots: *{max_bots}*\n\n"
                    f"*Build command:*\n"
                    f"`python scripts/deploy_node.py --tier {tier} "
                    f"--db-type {db_type} --llm-mode {llm_mode} "
                    f"--qdrant-mode {qdrant_mode} --industry {industry} "
                    f"--customer-name \"{full_name}\" "
                    f"--customer-email \"{email}\" "
                    f"--node-id {node_id} "
                    f"--generate-installer`"
                )
                try:
                    await query.message.get_bot().send_message(
                        chat_id=OWNER_ID,
                        text=owner_msg,
                        parse_mode="Markdown",
                    )
                except Exception as exc:
                    logger.error("Failed to notify owner of setup request: %s", exc)
            return


# ---------------------------------------------------------------------------
# Callback query handler (inline keyboard)
# ---------------------------------------------------------------------------

async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    query   = update.callback_query
    await query.answer()
    user_id = query.from_user.id
    data    = query.data or ""

    # Owner callbacks
    if data.startswith("owner:"):
        if not is_owner(user_id):
            await query.message.reply_text("Owner access only.")
            return
        sub = data[6:]
        if sub == "leads":
            await _owner_show_leads(query)
        elif sub == "queries":
            await _owner_show_queries(query)
        elif sub == "status":
            await _owner_show_status(query)
        elif sub == "profile":
            result = _execute_owner_action("my profile", user_id)
            if result:
                try:
                    await query.message.reply_text(result, parse_mode="Markdown")
                except Exception:
                    await query.message.reply_text(result)
        elif sub == "report":
            result = _execute_owner_action("daily report", user_id)
            if result:
                try:
                    await query.message.reply_text(result, parse_mode="Markdown")
                except Exception:
                    await query.message.reply_text(result)
        elif sub == "services":
            data_list = _dash_get("/api/services")
            if isinstance(data_list, list) and data_list:
                lines = ["*Your Services:*\n"]
                for svc in data_list:
                    name = svc.get("name", "?")
                    price = svc.get("price", 0)
                    currency = svc.get("currency", "USD")
                    active = "Active" if svc.get("is_active") else "Inactive"
                    lines.append(f"- *{name}* — {currency} {price:.2f} [{active}]")
                lines.append("\n_Manage in Dashboard → Services_")
                await query.message.reply_text("\n".join(lines), parse_mode="Markdown")
            else:
                await query.message.reply_text("No services configured. Add them in Dashboard → Services.")
        elif sub == "purchases":
            purchases = _dash_get("/api/license-purchases", limit=10)
            if isinstance(purchases, list) and purchases:
                lines = ["*Recent License Purchases:*\n"]
                for p in purchases[:10]:
                    ref = p.get("reference_no", "?")
                    ltype = p.get("license_type", "?")
                    status = p.get("status", "?")
                    email = p.get("email", "")
                    if "@" in email:
                        local, domain = email.split("@", 1)
                        email = f"{local[0]}***@{domain}" if len(local) > 1 else f"*@{domain}"
                    lines.append(f"`{ref}` — {ltype.upper()} [{status}] {email}")
                await query.message.reply_text("\n".join(lines), parse_mode="Markdown")
            else:
                await query.message.reply_text("No license purchases yet.")
        elif sub == "transactions":
            txns = _dash_get("/api/transactions", limit=10)
            if isinstance(txns, list) and txns:
                lines = ["*Recent Transactions:*\n"]
                for t in txns[:10]:
                    ref = t.get("reference_no", "?")
                    title = t.get("title", "?")
                    status = t.get("status", "?")
                    amount = t.get("amount", 0)
                    currency = t.get("currency", "USD")
                    lines.append(f"`{ref}` — {title} [{status}] {currency} {amount:.2f}")
                await query.message.reply_text("\n".join(lines), parse_mode="Markdown")
            else:
                await query.message.reply_text("No transactions yet.")
        elif sub == "affiliates":
            links = _dash_get("/api/affiliate-links", active_only=False)
            if isinstance(links, list) and links:
                lines = ["*Affiliate Links:*\n"]
                for l in links:
                    title = l.get("title", "?")
                    clicks = l.get("click_count", 0)
                    active = "Active" if l.get("is_active") else "Inactive"
                    lines.append(f"- *{title}* — {clicks} clicks [{active}]")
                lines.append("\n_Manage in Dashboard → Affiliate Links_")
                await query.message.reply_text("\n".join(lines), parse_mode="Markdown")
            else:
                await query.message.reply_text("No affiliate links. Add them in Dashboard → Affiliate Links.")
        elif sub == "sendemail":
            await query.message.reply_text(
                "*Send Email*\n\n"
                "Type in this format:\n"
                "`send email to user@example.com subject Hello body Your message here`",
                parse_mode="Markdown",
            )
        elif sub == "kb":
            try:
                r = requests.get(f"{QDRANT_URL}/collections/{QDRANT_COLLECTION}", timeout=3)
                mem_pts = r.json().get("result", {}).get("points_count", "?")
            except Exception:
                mem_pts = "?"
            try:
                r2 = requests.get(f"{QDRANT_URL}/collections/biznode_support", timeout=3)
                sup_pts = r2.json().get("result", {}).get("points_count", "?")
            except Exception:
                sup_pts = "?"
            await query.message.reply_text(
                f"*Knowledge Base*\n\n"
                f"Business KB: *{mem_pts}* entries\n"
                f"Support KB: *{sup_pts}* entries\n\n"
                f"_Manage in Dashboard → Knowledge Base_",
                parse_mode="Markdown",
            )
        elif sub == "custmenu":
            await query.message.reply_text(
                "Customer menu preview:", reply_markup=_main_menu()
            )
        return

    # ── Role selection callbacks ──────────────────────────────────────────────
    if data.startswith("role:"):
        new_role = data[5:]
        if new_role in VALID_PERSON_TYPES:
            update_bot_user(user_id, person_type=new_role)
            label = PERSON_TYPE_LABELS.get(new_role, new_role)
            await query.message.reply_text(
                f"Role updated to *{label}*.\n\nHere's your personalised menu:",
                parse_mode="Markdown",
                reply_markup=_main_menu(new_role),
            )
        else:
            await query.message.reply_text("Invalid role.")
        return

    # ── Info callbacks (role-specific content) ─────────────────────────────
    if data.startswith("info:"):
        sub = data[5:]
        await _handle_info_callback(query, user_id, sub)
        return

    # ── Transaction callbacks ─────────────────────────────────────────────
    if data.startswith("txn:"):
        sub = data[4:]
        await _handle_txn_callback(query, user_id, sub)
        return

    # ── Payment callbacks ──────────────────────────────────────────────────
    if data.startswith("pay:"):
        sub = data[4:]
        await _handle_pay_callback(query, user_id, sub)
        return

    # ── Buy service callbacks ──────────────────────────────────────────────
    if data.startswith("buy:"):
        service_id = data[4:]
        await _handle_buy_callback(query, user_id, service_id)
        return

    # ── Setup wizard callbacks ────────────────────────────────────────────
    if data.startswith("setup:"):
        sub = data[6:]
        await _handle_setup_callback(query, user_id, sub)
        return

    # ── License purchase callbacks ────────────────────────────────────────
    if data.startswith("license:"):
        sub = data[8:]
        await _handle_license_callback(query, user_id, sub)
        return

    # Customer / menu callbacks
    if data == "task:__ask":
        await query.message.reply_text(
            "Just type your question and I'll answer it from the knowledge base."
        )
        return

    if data.startswith("task:"):
        task_code = data[5:]
        prompt    = trigger_workflow(user_id, task_code)
        if prompt:
            await query.message.reply_text(prompt)
        else:
            await query.message.reply_text(
                f"No workflow set up for '{task_code}'. The owner can configure it in the dashboard."
            )


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    logger.info(
        "BizNode bot starting — OWNER_ID=%s DASHBOARD=%s MODEL=%s",
        OWNER_ID, DASHBOARD_URL, OLLAMA_MODEL,
    )
    ensure_qdrant_collection()
    _load_instruction()

    app = ApplicationBuilder().token(BOT_TOKEN).build()

    app.add_handler(CommandHandler("start",   cmd_start))
    app.add_handler(CommandHandler("help",    cmd_help))
    app.add_handler(CommandHandler("menu",    cmd_menu))
    app.add_handler(CommandHandler("report",  cmd_report))
    app.add_handler(CommandHandler("leads",   cmd_leads))
    app.add_handler(CommandHandler("queries", cmd_queries))
    app.add_handler(CommandHandler("status",  cmd_status))
    app.add_handler(CommandHandler("role",    cmd_role))
    app.add_handler(CommandHandler("clear",   cmd_clear))
    app.add_handler(CommandHandler("memory",  cmd_memory))

    # Transaction / Payment / Service commands — Pro-only gated
    if is_feature_allowed("crypto_service"):
        app.add_handler(CommandHandler("transact",       cmd_transact))
        app.add_handler(CommandHandler("pay",            cmd_pay))
        app.add_handler(CommandHandler("wallet",         cmd_wallet))
        app.add_handler(CommandHandler("verify_wallet",  cmd_verify_wallet))
    if is_feature_allowed("affiliate_service"):
        app.add_handler(CommandHandler("affiliate",      cmd_affiliate))

    # All-tier commands
    app.add_handler(CommandHandler("services",       cmd_services))
    app.add_handler(CommandHandler("negotiate",      cmd_negotiate))
    app.add_handler(CommandHandler("buy_license",    cmd_buy_license))
    app.add_handler(CommandHandler("verify_purchase", cmd_verify_purchase))
    app.add_handler(CommandHandler("purchase_status", cmd_purchase_status))

    # Owner tools — KB management & group posting
    app.add_handler(CommandHandler("kbadd",     cmd_kbadd))
    app.add_handler(CommandHandler("kbremove",  cmd_kbremove))
    app.add_handler(CommandHandler("postgroup", cmd_postgroup))
    app.add_handler(CommandHandler("grouppost", cmd_postgroup))  # alias

    if is_feature_allowed("bizbrain_service"):
        app.add_handler(CommandHandler("bizbrain",  cmd_bizbrain))
    if is_feature_allowed("bizchannel"):
        app.add_handler(CommandHandler("bo",        cmd_bo))

    app.add_handler(CommandHandler("setup",     cmd_setup))
    app.add_handler(CommandHandler("support",   cmd_support))
    app.add_handler(CommandHandler("version",   cmd_version))
    app.add_handler(CommandHandler("ceo",       cmd_ceo))
    app.add_handler(CommandHandler("founder",   cmd_ceo))  # alias
    app.add_handler(CommandHandler("sendupdate",  cmd_sendupdate))
    app.add_handler(CommandHandler("setpayload",  cmd_setpayload))
    app.add_handler(CommandHandler("applyupdate", cmd_applyupdate))

    # Fallback for gated Pro commands — show upgrade message instead of silence
    _gated_pro_cmds = [k for k, v in _PRO_COMMANDS.items() if not is_feature_allowed(v)]
    if _gated_pro_cmds:
        app.add_handler(CommandHandler(_gated_pro_cmds, _pro_upgrade_msg))
    app.add_handler(MessageHandler(
        filters.Document.PDF & filters.ChatType.PRIVATE,
        handle_update_packet_document,
    ))

    try:
        from videoautomation.telegram_handler import register_handlers as _reg_video
        _reg_video(app)
        logger.info("Video automation module loaded")
    except ImportError:
        logger.info("Video automation not available — /videotask disabled")

    # Group message handler — must be registered BEFORE private handler
    if TELEGRAM_GROUP_ID:
        logger.info("Group monitoring active for chat_id=%s", TELEGRAM_GROUP_ID)
        app.add_handler(
            MessageHandler(
                filters.Chat(chat_id=TELEGRAM_GROUP_ID) & filters.TEXT & ~filters.COMMAND,
                handle_group_message,
            )
        )

    app.add_handler(CallbackQueryHandler(handle_callback))
    app.add_handler(MessageHandler(
        filters.ChatType.PRIVATE & filters.TEXT & ~filters.COMMAND,
        handle_message,
    ))

    # Schedule group ad sender (every 6 hours)
    if TELEGRAM_GROUP_ID:
        app.job_queue.run_repeating(
            _send_pending_group_ads,
            interval=6 * 3600,
            first=60,
            name="group_ads",
        )
        logger.info("Group ad sender scheduled (every 6 hours)")

        # Daily business update — once every 24 hours (first run after 5 minutes)
        app.job_queue.run_repeating(
            _send_daily_business_update,
            interval=24 * 3600,
            first=300,
            name="daily_business_update",
        )
        logger.info("Daily business update scheduled (every 24 hours)")

    # Clear any stale getUpdates long-poll from a previous instance
    # to avoid "Conflict: terminated by other getUpdates request"
    import urllib.request
    try:
        urllib.request.urlopen(
            f"https://api.telegram.org/bot{BOT_TOKEN}/deleteWebhook?drop_pending_updates=true",
            timeout=10,
        )
        logger.info("Cleared stale Telegram polling session")
    except Exception as e:
        logger.warning(f"Could not clear stale session: {e}")

    app.run_polling(drop_pending_updates=True)
