server.py

from dotenv import load_dotenv
from pathlib import Path

ROOT_DIR = Path(__file__).parent
load_dotenv(ROOT_DIR / '.env')

import os
import io
import csv
import re
import uuid
import json
import asyncio
import logging
import base64
import secrets
from datetime import datetime, timezone, timedelta
from time import monotonic
from typing import List, Optional, Dict, Any

import httpx
import numpy as np
from openai import OpenAI

import firebase_admin
from firebase_admin import credentials, auth, firestore, storage
from google.cloud.firestore_v1.base_query import FieldFilter
from fastapi import FastAPI, APIRouter, HTTPException, Depends, Request, Response, UploadFile, File, Form, BackgroundTasks
from fastapi.responses import StreamingResponse
from fastapi.security import HTTPBearer
from starlette.middleware.cors import CORSMiddleware
from pydantic import BaseModel, EmailStr, Field, ConfigDict

# ----- Logger setup (must come before Firebase init so we can log errors) -----
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger("scatter")

# ----- Firebase init -----
if not firebase_admin._apps:
    firebase_json = os.environ.get("FIREBASE_JSON")
    firebase_json_path = os.environ.get("FIREBASE_JSON_PATH")
    cred = None
    if firebase_json:
        try:
            cred_dict = json.loads(firebase_json)
            cred = credentials.Certificate(cred_dict)
        except Exception as e:
            logger.error(f"Error parsing FIREBASE_JSON: {e}")
            cred = credentials.ApplicationDefault()
    elif firebase_json_path:
        try:
            cred = credentials.Certificate(firebase_json_path)
        except Exception as e:
            logger.error(f"Error loading FIREBASE_JSON_PATH ({firebase_json_path}): {e}")
            cred = credentials.ApplicationDefault()
    else:
        cred = credentials.ApplicationDefault()

    firebase_admin.initialize_app(cred, {
        'projectId': os.environ.get("FIREBASE_PROJECT_ID", "scatter-studio-live-2026"),
        'storageBucket': os.environ.get("FIREBASE_STORAGE_BUCKET", "scatter-studio-live-2026.firebasestorage.app")
    })

db = firestore.client()
bucket = storage.bucket()

app = FastAPI(title="Scatter Studio API")

# CORS: auth is via `Authorization: Bearer` (no cookies), so credential-less
# wildcard is safe and needs no proxy — the Vercel frontend calls the HF Space
# API directly. We still let ALLOWED_ORIGINS pin specific origins in production
# (comma-separated). Default includes the Vercel domain + local dev. Set
# ALLOWED_ORIGINS="*" to allow any origin.
_default_origins = [
    "https://scatter-studio-eight.vercel.app",
    "http://localhost:3000",
    "http://127.0.0.1:3000",
]
_env_origins = (os.environ.get("ALLOWED_ORIGINS") or "").strip()
if _env_origins == "*":
    _cors_origins = ["*"]
elif _env_origins:
    _cors_origins = [o.strip() for o in _env_origins.split(",") if o.strip()]
else:
    # No override set → allow the known origins PLUS any *.vercel.app preview
    # deploy via regex, so branch/preview URLs work without reconfiguring.
    _cors_origins = _default_origins

app.add_middleware(
    CORSMiddleware,
    allow_origins=_cors_origins,
    allow_origin_regex=r"https://.*\.vercel\.app" if _cors_origins != ["*"] else None,
    allow_credentials=False,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["*"],
)

# Ensure CORS headers are present on unhandled 500s too
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    logger.error(f"Unhandled exception: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"},
        headers={"Access-Control-Allow-Origin": "*"},
    )

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
        headers={"Access-Control-Allow-Origin": "*"},
    )

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    logger.error(f"Validation error: {exc}")
    return JSONResponse(
        status_code=422,
        content={"detail": str(exc)},
        headers={"Access-Control-Allow-Origin": "*"},
    )

api_router = APIRouter(prefix="/api")
security = HTTPBearer(auto_error=False)

_CACHE: Dict[str, tuple[float, Any]] = {}


def _cache_get(key: str):
    hit = _CACHE.get(key)
    if not hit:
        return None
    expires_at, value = hit
    if expires_at <= monotonic():
        _CACHE.pop(key, None)
        return None
    return value


def _cache_set(key: str, value: Any, ttl: int = 20):
    _CACHE[key] = (monotonic() + ttl, value)
    return value


def _cache_invalidate_user(uid: str, *areas: str):
    prefixes = tuple(f"{area}:{uid}" for area in areas) if areas else (f"user:{uid}",)
    for key in list(_CACHE):
        if key.startswith(prefixes):
            _CACHE.pop(key, None)


# Stale-while-revalidate store: keeps the LAST computed value forever (until
# overwritten) so a cache MISS can still return instantly with slightly-stale
# data while a fresh value is recomputed in the background. This is the biggest
# perceived-speed win on the dashboard — the user never waits on a cold scan.
_STALE: Dict[str, Any] = {}
_INFLIGHT: set = set()


def _swr(key: str, builder, fresh_ttl: int = 60):
    """Return `builder()`'s result with stale-while-revalidate semantics.
    - Fresh cache hit  → return it.
    - Stale value present → return it NOW, refresh in a background thread.
    - Nothing at all   → compute synchronously (first-ever load) and store."""
    fresh = _cache_get(key)
    if fresh is not None:
        return fresh

    def _refresh():
        try:
            val = builder()
            _cache_set(key, val, ttl=fresh_ttl)
            _STALE[key] = val
        except Exception as e:
            logger.warning(f"[SWR] refresh failed for {key}: {e}")
        finally:
            _INFLIGHT.discard(key)

    stale = _STALE.get(key)
    if stale is not None:
        if key not in _INFLIGHT:
            _INFLIGHT.add(key)
            import threading as _th
            _th.Thread(target=_refresh, daemon=True).start()
        return stale

    # Cold: compute once synchronously.
    val = builder()
    _cache_set(key, val, ttl=fresh_ttl)
    _STALE[key] = val
    return val


# ----- Helpers -----
import hashlib as _hashlib

# Short-lived token->decoded cache. auth.verify_id_token() is a blocking network
# round-trip to Google's cert endpoint + JWT verify on EVERY request — the single
# biggest per-request latency. Caching the verified result per token for a short
# window (well under the token's own ~1h validity) makes repeat requests skip it
# entirely. Keyed by a hash of the token so we never store the raw JWT as a key.
_TOKEN_TTL_SEC = 120


def _verify_token_cached(token: str) -> dict:
    tkey = "tok:" + _hashlib.sha256(token.encode()).hexdigest()[:32]
    hit = _cache_get(tkey)
    if hit:
        return hit
    decoded = auth.verify_id_token(token)
    # Cache only until the min of our TTL and the token's own expiry.
    import time as _t
    ttl = _TOKEN_TTL_SEC
    exp = decoded.get("exp")
    if exp:
        ttl = max(5, min(_TOKEN_TTL_SEC, int(exp - _t.time() - 10)))
    _cache_set(tkey, decoded, ttl=ttl)
    return decoded


async def get_current_user(request: Request) -> dict:
    token = None
    auth_header = request.headers.get("Authorization", "")
    if auth_header.startswith("Bearer "):
        token = auth_header[7:]
    if not token:
        raise HTTPException(status_code=401, detail="Not authenticated")
    try:
        # Run the (blocking) verify + Firestore read off the event loop so one
        # slow auth check can't stall every other concurrent request.
        decoded_token = await asyncio.to_thread(_verify_token_cached, token)
        uid = decoded_token['uid']
        cached_user = _cache_get(f"user:{uid}")
        if cached_user:
            return cached_user
        user_doc = await asyncio.to_thread(lambda: db.collection("users").document(uid).get())
        if not user_doc.exists:
            user_data = {
                "id": uid,
                "email": decoded_token.get("email"),
                "name": decoded_token.get("name", "User"),
                "role": "user",
                "workspace": "Scatter Studio",
                "created_at": datetime.now(timezone.utc).isoformat(),
                "api_key": "sk-scatter-" + secrets.token_hex(20),
            }
            db.collection("users").document(uid).set(user_data)
            return _cache_set(f"user:{uid}", user_data, ttl=600)
        data = user_doc.to_dict()
        # Backfill api_key if missing
        if not data.get("api_key"):
            data["api_key"] = "sk-scatter-" + secrets.token_hex(20)
            db.collection("users").document(uid).update({"api_key": data["api_key"]})
        return _cache_set(f"user:{uid}", data, ttl=600)
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Auth error: {e}")
        raise HTTPException(status_code=401, detail="Invalid or expired token")


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


# ----- Models -----
class UserPublic(BaseModel):
    model_config = ConfigDict(extra="ignore")
    id: str
    email: EmailStr
    name: str
    role: str = "user"
    workspace: str = "Scatter Studio"
    created_at: str
    api_key: Optional[str] = None


class AgentIn(BaseModel):
    model_config = ConfigDict(extra="ignore")
    name: str
    role: str = "Sales"
    persona: str = "You are a friendly customer support agent."
    welcome_message: str = "Hello! How can I help you today?"
    language: str = "Hinglish"
    voice_provider: str = "Sarvam"
    voice_id: str = "shreya"
    voice_speed: float = 1.0
    voice_pitch: float = 0.0
    llm_provider: str = "Groq"
    llm_model: str = "llama-3.3-70b-versatile"
    stt_provider: str = "Sarvam"
    knowledge_base_ids: List[str] = []
    phone_number: Optional[str] = None
    status: str = "active"
    # Silence watchdog (read by the LiveKit worker): 0 disables that stage.
    silence_nudge_seconds: int = 15
    silence_disconnect_seconds: int = 30
    silence_nudge_message: str = ""
    silence_goodbye_message: str = ""
    # Per-agent post-call integration allow-list (default-ON; only False disables).
    integrations: Dict[str, bool] = Field(default_factory=dict)
    # Auto-created Google Sheet CRM id (for users without HubSpot/Zoho).
    crm_spreadsheet_id: str = ""


class Agent(AgentIn):
    id: str
    created_at: str
    updated_at: str
    calls_handled: int = 0
    avg_duration: float = 0.0
    last_active: Optional[str] = None


class CampaignIn(BaseModel):
    model_config = ConfigDict(extra="ignore")
    name: str
    agent_id: str
    pacing_per_minute: int = 10
    schedule: Optional[str] = None
    contacts: List[Dict[str, Any]] = []


class Campaign(CampaignIn):
    id: str
    status: str = "draft"
    created_at: str
    progress: Dict[str, int] = Field(default_factory=lambda: {
        "total": 0, "dialed": 0, "connected": 0, "completed": 0, "failed": 0, "leads": 0,
    })


class CallLog(BaseModel):
    model_config = ConfigDict(extra="ignore")
    id: str
    agent_id: Optional[str] = ""
    agent_name: str = ""
    direction: str = "inbound"
    phone_number: str = ""
    contact_name: Optional[str] = None
    duration_seconds: int = 0
    status: str = "queued"
    sentiment: str = "neutral"
    lead_tag: str = "none"
    intent: str = ""
    summary: str = ""
    analysis: str = ""
    topics: List[str] = []
    transcript: List[Dict[str, Any]] = []
    started_at: str = ""
    recording_url: Optional[str] = None
    provider_call_id: Optional[str] = None
    channel: str = ""
    # Lead fields surfaced per-call for the Call Analytics dashboard.
    lead_name: str = ""
    lead_email: str = ""
    lead_company: str = ""
    lead_phone: str = ""
    lead_score: int = 0
    # Post-call automation trail ({type,label,status,detail}).
    actions: List[Dict[str, Any]] = []
    # True while the agent worker is live-syncing this call (Command Center).
    live: bool = False


class Lead(BaseModel):
    model_config = ConfigDict(extra="ignore")
    id: str = ""
    agent_id: str = ""
    agent_name: str = ""
    name: str = ""
    email: str = ""
    company: str = ""
    phone: str = ""
    intent: str = ""
    message: str = ""
    source: str = ""
    status: str = "new"
    sentiment: str = "neutral"
    score: int = 0


class PhoneNumber(BaseModel):
    id: str
    number: str
    provider: str = "Vobiz"
    label: str
    assigned_agent_id: Optional[str] = None
    status: str = "active"
    sip_domain: Optional[str] = None
    sip_username: Optional[str] = None
    sip_password: Optional[str] = None


class KnowledgeDoc(BaseModel):
    model_config = ConfigDict(extra="ignore")
    id: str
    name: str = ""
    size_kb: float = 0
    pages: int = 0
    status: str = "ready"
    uploaded_at: str = ""
    chunks: Optional[int] = 0


# ----- Auth Routes -----
@api_router.get("/auth/me", response_model=UserPublic)
async def me(user: dict = Depends(get_current_user)):
    return UserPublic(**user)


# ----- LiveKit -----
from livekit import api as livekit_api


@api_router.get("/livekit/token")
async def get_livekit_token(agent_id: str, user: dict = Depends(get_current_user)):
    agent_doc = db.collection("agents").document(agent_id).get()
    if not agent_doc.exists or agent_doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=404, detail="Agent not found")

    # Include agent_id + owner_id in the room name so the worker can load config reliably.
    room_name = f"agent_{agent_id}__user_{user['id']}__session_{uuid.uuid4().hex}"
    participant_name = user.get("name") or "User"

    api_key = os.environ.get("LIVEKIT_API_KEY")
    api_secret = os.environ.get("LIVEKIT_API_SECRET")
    livekit_url = os.environ.get("LIVEKIT_URL")
    if not api_key or not api_secret:
        raise HTTPException(status_code=500, detail="LIVEKIT_API_KEY/SECRET not configured")
    if not livekit_url:
        raise HTTPException(status_code=500, detail="LIVEKIT_URL not configured")

    metadata = json.dumps({"agent_id": agent_id, "owner_id": user["id"]})
    agent_dispatch_name = os.environ.get("LIVEKIT_AGENT_NAME", "scatterstudio-agent")

    token = livekit_api.AccessToken(api_key, api_secret)
    token = (
        token.with_identity(user["id"])
        .with_name(participant_name)
        .with_metadata(metadata)
        .with_grants(
            livekit_api.VideoGrants(
                room_join=True,
                room=room_name,
                can_publish=True,
                can_subscribe=True,
            )
        )
    )
    lkapi = None
    try:
        lkapi = livekit_api.LiveKitAPI(url=livekit_url, api_key=api_key, api_secret=api_secret)
        await lkapi.agent_dispatch.create_dispatch(
            livekit_api.CreateAgentDispatchRequest(
                room=room_name,
                agent_name=agent_dispatch_name,
                metadata=metadata,
            )
        )
    except Exception as e:
        logger.error(f"LiveKit agent dispatch failed for room {room_name}: {e}")
        raise HTTPException(status_code=502, detail="Failed to dispatch LiveKit agent")
    finally:
        if lkapi:
            await lkapi.aclose()
    return {"token": token.to_jwt(), "url": livekit_url, "room": room_name, "agent_id": agent_id}


# ----- LiveKit SIP helpers -----

def _lk_creds():
    """Return (api_key, api_secret, livekit_url) or raise."""
    api_key = os.environ.get("LIVEKIT_API_KEY", "")
    api_secret = os.environ.get("LIVEKIT_API_SECRET", "")
    url = os.environ.get("LIVEKIT_URL", "")
    if not (api_key and api_secret and url):
        raise HTTPException(status_code=500, detail="LIVEKIT_API_KEY/SECRET/URL not configured")
    return api_key, api_secret, url


def _to_e164(num: str, default_cc: str = "91") -> str:
    """Normalise a phone number to E.164 for the LiveKit→SIP outbound leg.
    A bare 10-digit Indian number is rejected by the trunk (the call never rings
    even though CreateSIPParticipant returns OK), so we always add +<cc>.
    Indian-default: 10-digit → +91XXXXXXXXXX; 12-digit starting 91 → +91…;
    already-+ → unchanged."""
    s = str(num or "").strip().replace(" ", "").replace("-", "")
    if not s:
        return s
    if s.startswith("+"):
        return s
    s = s.lstrip("0")
    if len(s) == 10:
        return f"+{default_cc}{s}"
    if s.startswith(default_cc) and len(s) >= (len(default_cc) + 10):
        return f"+{s}"
    return f"+{s}"


# Region / phone-prefix → language mapping
_PHONE_LANG_MAP = [
    # India regional prefixes (mobile series)
    # Hindi belt
    ("+9151", "hi-IN"), ("+9152", "hi-IN"), ("+9153", "hi-IN"),
    ("+9155", "hi-IN"), ("+9156", "hi-IN"), ("+9157", "hi-IN"),
    ("+9161", "hi-IN"), ("+9162", "hi-IN"), ("+9163", "hi-IN"),
    # Tamil Nadu
    ("+9144", "ta-IN"), ("+9142", "ta-IN"), ("+9143", "ta-IN"),
    ("+9145", "ta-IN"),
    # Andhra / Telangana
    ("+9140", "te-IN"), ("+9186", "te-IN"), ("+9187", "te-IN"),
    ("+9188", "te-IN"),
    # Karnataka
    ("+9180", "kn-IN"), ("+9182", "kn-IN"),
    # Maharashtra
    ("+9122", "mr-IN"), ("+9120", "mr-IN"), ("+9121", "mr-IN"),
    # Gujarat
    ("+9179", "gu-IN"), ("+9126", "gu-IN"), ("+9127", "gu-IN"),
    # Bengal
    ("+9133", "bn-IN"), ("+9134", "bn-IN"),
    # Punjab
    ("+9117", "pa-IN"), ("+9118", "pa-IN"),
    # Kerala
    ("+9147", "ml-IN"), ("+9148", "ml-IN"),
    # Generic India fallback
    ("+91", "hi-IN"),
]


def _detect_language_from_phone(phone: str) -> str:
    """Detect likely language from phone number prefix."""
    phone = re.sub(r"[\s\-\(\)]", "", phone)
    if not phone.startswith("+"):
        if phone.startswith("0"):
            phone = "+91" + phone[1:]
        elif len(phone) == 10:
            phone = "+91" + phone
    for prefix, lang in _PHONE_LANG_MAP:
        if phone.startswith(prefix):
            return lang
    return "en-IN"


_LANG_CODE_TO_NAME = {
    "hi-IN": "Hindi", "en-IN": "English", "ta-IN": "Tamil",
    "te-IN": "Telugu", "kn-IN": "Kannada", "mr-IN": "Marathi",
    "gu-IN": "Gujarati", "bn-IN": "Bengali", "pa-IN": "Punjabi",
    "ml-IN": "Malayalam",
}


async def _create_sip_outbound_call(
    agent_id: str, owner_id: str, to_number: str, from_number: str, call_id: str,
) -> Dict[str, Any]:
    """Create a LiveKit room with an agent, then dial out via SIP participant.

    This bridges the AI voice agent (in LiveKit) to a real phone call (via SIP).
    """
    api_key, api_secret, livekit_url = _lk_creds()

    room_name = f"agent_{agent_id}__user_{owner_id}__sip_{call_id}"
    agent_dispatch_name = os.environ.get("LIVEKIT_AGENT_NAME", "scatterstudio-agent")

    # Detect language from the callee's phone number
    detected_lang = _detect_language_from_phone(to_number)

    # Pre-call CRM enrichment: if the owner has a connected CRM and the callee is
    # a known contact, pass a hint so the agent greets them by name. Soft-fail.
    crm_hint = ""
    try:
        import integrations as _int
        crm = _int.execute_tool(owner_id, "lookup_crm_contact", {"phone": to_number})
        if (crm or {}).get("ok") and crm.get("contact"):
            c = crm["contact"]
            nm = (c.get("firstname") or c.get("First_Name") or "").strip()
            co = (c.get("company") or c.get("Account_Name") or "").strip()
            parts = [f"{nm}".strip(), co]
            summary = " from ".join(p for p in parts if p)
            if summary:
                crm_hint = summary
    except Exception as e:
        logger.warning(f"Pre-call CRM lookup soft-fail: {e}")

    metadata = json.dumps({
        "agent_id": agent_id,
        "owner_id": owner_id,
        "call_id": call_id,
        "direction": "outbound",
        "phone_number": to_number,
        "detected_language": detected_lang,
        "crm_hint": crm_hint,
    })

    lkapi = livekit_api.LiveKitAPI(url=livekit_url, api_key=api_key, api_secret=api_secret)
    try:
        # 1. Dispatch agent to the room
        await lkapi.agent_dispatch.create_dispatch(
            livekit_api.CreateAgentDispatchRequest(
                room=room_name,
                agent_name=agent_dispatch_name,
                metadata=metadata,
            )
        )

        # 2. Resolve SIP trunk for outbound — look for user's configured trunk first,
        #    then fall back to the platform default trunk
        sip_trunk_id = os.environ.get("LIVEKIT_SIP_TRUNK_ID", "")

        # Check if user has a phone number with SIP credentials for this agent
        phone_docs = list(
            db.collection("phone_numbers")
            .where(filter=FieldFilter("owner_id", "==", owner_id))
            .where(filter=FieldFilter("assigned_agent_id", "==", agent_id))
            .limit(1)
            .stream()
        )
        phone_data = phone_docs[0].to_dict() if phone_docs else None

        if not sip_trunk_id and phone_data and phone_data.get("sip_trunk_id"):
            sip_trunk_id = phone_data["sip_trunk_id"]

        if not sip_trunk_id:
            # Auto-create an outbound trunk using Vobiz env vars or phone_data SIP creds
            sip_domain = ""
            sip_user = ""
            sip_pass = ""
            if phone_data:
                sip_domain = phone_data.get("sip_domain") or ""
                sip_user = phone_data.get("sip_username") or ""
                sip_pass = phone_data.get("sip_password") or ""
            if not sip_domain:
                sip_domain = os.environ.get("VOBIZ_SIP_DOMAIN", "")
            if not sip_user:
                sip_user = os.environ.get("VOBIZ_SIP_USERNAME", os.environ.get("VOBIZ_AUTH_ID", ""))
            if not sip_pass:
                sip_pass = os.environ.get("VOBIZ_SIP_PASSWORD", os.environ.get("VOBIZ_AUTH_SECRET", ""))

            if not sip_domain:
                return {"ok": False, "error": "No SIP trunk configured. Add SIP domain in Phone Numbers or set VOBIZ_SIP_DOMAIN."}

            # Create outbound trunk on-the-fly
            trunk = await lkapi.sip.create_sip_outbound_trunk(
                livekit_api.CreateSIPOutboundTrunkRequest(
                    trunk=livekit_api.SIPOutboundTrunkInfo(
                        name=f"scatter-outbound-{agent_id[:8]}",
                        address=sip_domain,
                        numbers=[from_number],
                        auth_username=sip_user,
                        auth_password=sip_pass,
                    )
                )
            )
            sip_trunk_id = trunk.sip_trunk_id

            # Persist trunk ID so we don't recreate it
            if phone_data:
                db.collection("phone_numbers").document(phone_data["id"]).update({
                    "sip_trunk_id": sip_trunk_id
                })

        # 3. Create SIP participant — this dials the phone number.
        # With a pre-made / auto-created OUTBOUND trunk, LiveKit already knows the
        # SIP address (the trunk's `address`), so we pass the RAW E.164 number as
        # sip_call_to. Building a `sip:num@domain` URI here fights the trunk config
        # and the callee never rings. Only fall back to a URI if the caller
        # explicitly passed one. Also normalise bare 10-digit numbers to E.164 —
        # the trunk rejects non-E.164 silently.
        if to_number.startswith("sip:"):
            sip_call_to = to_number
        else:
            sip_call_to = _to_e164(to_number)

        participant = await lkapi.sip.create_sip_participant(
            livekit_api.CreateSIPParticipantRequest(
                sip_trunk_id=sip_trunk_id,
                sip_call_to=sip_call_to,
                room_name=room_name,
                participant_identity=f"phone_{to_number}",
                participant_name=f"Phone {to_number}",
                participant_metadata=json.dumps({"phone": to_number, "call_id": call_id}),
                wait_until_answered=False,
            )
        )

        logger.info(f"SIP outbound started: room={room_name} trunk={sip_trunk_id} to={sip_call_to}")
        return {"ok": True, "room_name": room_name, "sip_trunk_id": sip_trunk_id, "participant_id": participant.participant_identity}

    except Exception as e:
        logger.error(f"SIP outbound call failed: {type(e).__name__}: {e}")
        return {"ok": False, "error": str(e)}
    finally:
        await lkapi.aclose()


# ----- SIP Trunk management endpoints -----

class SIPTrunkIn(BaseModel):
    name: str = ""
    sip_domain: str
    sip_username: str = ""
    sip_password: str = ""
    numbers: List[str] = []
    direction: str = "outbound"  # "inbound" or "outbound"
    agent_id: Optional[str] = None


@api_router.post("/sip/trunks")
async def create_sip_trunk(payload: SIPTrunkIn, user: dict = Depends(get_current_user)):
    """Create a LiveKit SIP trunk (inbound or outbound) and persist it."""
    api_key, api_secret, livekit_url = _lk_creds()
    lkapi = livekit_api.LiveKitAPI(url=livekit_url, api_key=api_key, api_secret=api_secret)
    agent_dispatch_name = os.environ.get("LIVEKIT_AGENT_NAME", "scatterstudio-agent")

    try:
        if payload.direction == "inbound":
            trunk = await lkapi.sip.create_sip_inbound_trunk(
                livekit_api.CreateSIPInboundTrunkRequest(
                    trunk=livekit_api.SIPInboundTrunkInfo(
                        name=payload.name or f"scatter-inbound-{user['id'][:8]}",
                        numbers=payload.numbers,
                        auth_username=payload.sip_username or "",
                        auth_password=payload.sip_password or "",
                    )
                )
            )

            # Create dispatch rule so inbound calls auto-route to an agent room
            if payload.agent_id:
                metadata = json.dumps({
                    "agent_id": payload.agent_id,
                    "owner_id": user["id"],
                    "direction": "inbound",
                })
                await lkapi.sip.create_sip_dispatch_rule(
                    livekit_api.CreateSIPDispatchRuleRequest(
                        trunk_ids=[trunk.sip_trunk_id],
                        rule=livekit_api.SIPDispatchRule(
                            dispatch_rule_individual=livekit_api.SIPDispatchRuleIndividual(
                                room_prefix=f"agent_{payload.agent_id}__user_{user['id']}__sip_",
                                pin="",
                            )
                        ),
                        metadata=metadata,
                    )
                )
        else:
            trunk = await lkapi.sip.create_sip_outbound_trunk(
                livekit_api.CreateSIPOutboundTrunkRequest(
                    trunk=livekit_api.SIPOutboundTrunkInfo(
                        name=payload.name or f"scatter-outbound-{user['id'][:8]}",
                        address=payload.sip_domain,
                        numbers=payload.numbers,
                        auth_username=payload.sip_username or "",
                        auth_password=payload.sip_password or "",
                    )
                )
            )

        # Store trunk info in Firestore
        trunk_data = {
            "id": trunk.sip_trunk_id,
            "name": payload.name or trunk.name,
            "direction": payload.direction,
            "sip_domain": payload.sip_domain,
            "numbers": payload.numbers,
            "agent_id": payload.agent_id,
            "owner_id": user["id"],
            "created_at": now_iso(),
        }
        db.collection("sip_trunks").document(trunk.sip_trunk_id).set(trunk_data)

        return {"ok": True, "trunk_id": trunk.sip_trunk_id, **trunk_data}
    except Exception as e:
        logger.error(f"SIP trunk creation failed: {e}")
        raise HTTPException(status_code=500, detail=f"SIP trunk creation failed: {e}")
    finally:
        await lkapi.aclose()


@api_router.get("/sip/trunks")
async def list_sip_trunks(user: dict = Depends(get_current_user)):
    """List SIP trunks for the current user."""
    docs = db.collection("sip_trunks").where(filter=FieldFilter("owner_id", "==", user["id"])).stream()
    return [doc.to_dict() for doc in docs]


@api_router.delete("/sip/trunks/{trunk_id}")
async def delete_sip_trunk(trunk_id: str, user: dict = Depends(get_current_user)):
    """Delete a SIP trunk."""
    doc = db.collection("sip_trunks").document(trunk_id).get()
    if not doc.exists or doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=404, detail="Trunk not found")
    api_key, api_secret, livekit_url = _lk_creds()
    lkapi = livekit_api.LiveKitAPI(url=livekit_url, api_key=api_key, api_secret=api_secret)
    try:
        await lkapi.sip.delete_sip_trunk(
            livekit_api.DeleteSIPTrunkRequest(sip_trunk_id=trunk_id)
        )
    except Exception as e:
        logger.warning(f"LiveKit SIP trunk delete failed (may already be gone): {e}")
    finally:
        await lkapi.aclose()
    db.collection("sip_trunks").document(trunk_id).delete()
    return {"ok": True}


# ----- Agents -----
@api_router.get("/agents", response_model=List[Agent])
def list_agents(user: dict = Depends(get_current_user)):
    cached = _cache_get(f"agents:{user['id']}")
    if cached is not None:
        return cached
    docs = db.collection("agents").where(filter=FieldFilter("owner_id", "==", user["id"])).stream()
    return _cache_set(f"agents:{user['id']}", [doc.to_dict() for doc in docs], ttl=20)


@api_router.post("/agents", response_model=Agent)
def create_agent(payload: AgentIn, user: dict = Depends(get_current_user)):
    if not payload.name.strip():
        raise HTTPException(status_code=400, detail="Agent name is required")
    aid = str(uuid.uuid4())
    doc_data = {
        "id": aid,
        **payload.model_dump(),
        "owner_id": user["id"],
        "created_at": now_iso(),
        "updated_at": now_iso(),
        "calls_handled": 0,
        "avg_duration": 0.0,
        "last_active": None,
    }
    db.collection("agents").document(aid).set(doc_data)
    _cache_invalidate_user(user["id"], "agents", "analytics", "leaderboard")
    return {k: v for k, v in doc_data.items() if k != "owner_id"}


@api_router.get("/agents/{agent_id}", response_model=Agent)
def get_agent(agent_id: str, user: dict = Depends(get_current_user)):
    agent_doc = db.collection("agents").document(agent_id).get()
    if not agent_doc.exists or agent_doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=404, detail="Agent not found")
    return agent_doc.to_dict()


# ----- ScatterAI: build an agent from a natural-language chat -----
class BuilderMsg(BaseModel):
    messages: List[Dict[str, Any]] = []   # [{role: 'user'|'assistant', content: str}]


_BUILDER_SYSTEM = (
    "You are ScatterAI, an expert assistant that helps a user design a voice AI agent by chatting. "
    "Ask brief, friendly questions to learn: the agent's PURPOSE (sales/support/reception/etc.), the "
    "BUSINESS it represents, its NAME, TONE, LANGUAGE, and the GREETING it should open with. Keep replies "
    "short and conversational — ONE or TWO questions at a time.\n\n"
    "When you have ENOUGH to build a great agent (usually after 2-4 exchanges, or immediately if the user "
    "gave a full description), output the final agent as a JSON object on its OWN line, wrapped EXACTLY like:\n"
    "<AGENT>{...}</AGENT>\n"
    "The JSON MUST have these keys: name, role (one of Sales|Support|Reception|Outbound|Survey), language "
    "(one of Hinglish|Hindi|English|Tamil|Telugu|Kannada|Marathi|Gujarati|Bengali|Punjabi|Malayalam), "
    "persona (a detailed, vivid system prompt describing who the agent is, the business, what it does, its "
    "tone, and how it should behave — write this richly, 3-6 sentences), welcome_message (the exact first "
    "line it speaks), voice_id (pick a Sarvam voice: shreya/simran/priya/neha for female, shubh/aditya/"
    "manan/rahul for male). Before the <AGENT> block, write ONE friendly sentence like 'Here's your agent — "
    "review and create it!'. Do NOT include the JSON if you still need info; ask a question instead."
)


@api_router.post("/agents/build")
async def scatterai_build(payload: BuilderMsg, user: dict = Depends(get_current_user)):
    """ScatterAI chat: the user describes the agent they want; NVIDIA NIM drives
    the conversation and, when ready, emits a full agent config (<AGENT>{...}</AGENT>)
    the frontend can one-click create. Falls back to Groq if NVIDIA is unset."""
    msgs = [{"role": "system", "content": _BUILDER_SYSTEM}]
    for m in (payload.messages or [])[-12:]:
        r = m.get("role")
        if r in ("user", "assistant") and m.get("content"):
            msgs.append({"role": r, "content": str(m["content"])[:4000]})
    if len(msgs) == 1:
        msgs.append({"role": "user", "content": "Hi! Help me build a voice agent."})

    nvidia_key = os.environ.get("NVIDIA_API_KEY")
    groq_key = os.environ.get("GROQ_API_KEY")
    providers = []
    if nvidia_key:
        providers.append(("https://integrate.api.nvidia.com/v1", nvidia_key,
                          os.environ.get("BUILDER_MODEL", "nvidia/nemotron-3-ultra-550b-a55b"), True))
    if groq_key:
        providers.append(("https://api.groq.com/openai/v1", groq_key,
                          os.environ.get("BUILDER_GROQ_MODEL", "openai/gpt-oss-120b"), False))
    if not providers:
        raise HTTPException(status_code=500, detail="No LLM configured (set NVIDIA_API_KEY or GROQ_API_KEY)")

    reply = ""
    for base_url, key, model, is_nv in providers:
        try:
            body = {"model": model, "messages": msgs, "temperature": 0.6, "max_tokens": 900}
            if is_nv and model.startswith(("nvidia/nemotron", "z-ai/glm")):
                body["chat_template_kwargs"] = {"enable_thinking": False}
            async with httpx.AsyncClient(timeout=40.0) as client:
                r = await client.post(f"{base_url}/chat/completions",
                                      headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
                                      json=body)
                r.raise_for_status()
                reply = r.json()["choices"][0]["message"]["content"]
            break
        except Exception as e:
            logger.warning(f"[ScatterAI] provider failed: {e}")
            reply = ""

    if not reply:
        return {"reply": "Sorry, I couldn't reach the model right now. Please try again.", "agent": None}

    # Extract the agent JSON if the model emitted one.
    agent = None
    m = re.search(r"<AGENT>\s*(\{.*?\})\s*</AGENT>", reply, re.DOTALL)
    if m:
        try:
            raw = json.loads(m.group(1))
            # Normalise into a valid AgentIn payload with sane defaults.
            agent = {
                "name": (raw.get("name") or "My Agent")[:100],
                "role": raw.get("role") if raw.get("role") in ("Sales", "Support", "Reception", "Outbound", "Survey") else "Sales",
                "language": raw.get("language") or "Hinglish",
                "persona": raw.get("persona") or "You are a helpful voice agent.",
                "welcome_message": raw.get("welcome_message") or "Hello! How can I help you today?",
                "voice_provider": "Sarvam",
                "voice_id": raw.get("voice_id") or "shreya",
                "llm_provider": "Gemini",
                # 3.1 speaks + transcribes fine now that the worker passes
                # output_audio_transcription=None (that arg, not the model, caused 1007).
                "llm_model": "gemini-3.1-flash-live-preview",
            }
        except Exception as e:
            logger.warning(f"[ScatterAI] agent JSON parse failed: {e}")
            agent = None
    # Show the human-facing text (strip the <AGENT> block from the reply).
    clean_reply = re.sub(r"<AGENT>.*?</AGENT>", "", reply, flags=re.DOTALL).strip()
    return {"reply": clean_reply or "Here's your agent — review and create it!", "agent": agent}


@api_router.get("/agents/{agent_id}/sheet")
def get_agent_sheet(agent_id: str, user: dict = Depends(get_current_user)):
    """Return the agent's auto-created Google Sheet CRM link + a preview of the
    latest rows, so the user can view captured caller data from inside the app."""
    agent_doc = db.collection("agents").document(agent_id).get()
    if not agent_doc.exists or agent_doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=404, detail="Agent not found")
    sid = (agent_doc.to_dict() or {}).get("crm_spreadsheet_id") or ""
    if not sid:
        return {"connected": False, "sheet_url": None, "rows": [],
                "message": "No CRM sheet yet — connect Google Sheets and take a call; the agent auto-creates one."}
    url = f"https://docs.google.com/spreadsheets/d/{sid}/edit"
    rows = []
    try:
        import integrations as _int
        res = _int.execute_tool(user["id"], "read_sheet_rows", {"spreadsheet_id": sid, "sheet_name": "CRM"})
        if (res or {}).get("ok"):
            rows = (res.get("rows") or [])[:50]
    except Exception as e:
        logger.warning(f"[Sheet] preview read failed: {e}")
    return {"connected": True, "sheet_url": url, "spreadsheet_id": sid, "rows": rows}


@api_router.put("/agents/{agent_id}", response_model=Agent)
def update_agent(agent_id: str, payload: AgentIn, user: dict = Depends(get_current_user)):
    agent_ref = db.collection("agents").document(agent_id)
    agent_doc = agent_ref.get()
    if not agent_doc.exists or agent_doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=404, detail="Agent not found")
    update_data = {**payload.model_dump(), "updated_at": now_iso()}
    agent_ref.update(update_data)
    _cache_invalidate_user(user["id"], "agents", "analytics", "leaderboard")
    return agent_ref.get().to_dict()


@api_router.delete("/agents/{agent_id}")
def delete_agent(agent_id: str, user: dict = Depends(get_current_user)):
    agent_ref = db.collection("agents").document(agent_id)
    agent_doc = agent_ref.get()
    if not agent_doc.exists or agent_doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=404, detail="Agent not found")
    agent_ref.delete()
    _cache_invalidate_user(user["id"], "agents", "analytics", "leaderboard")
    return {"ok": True}


class TestMsgIn(BaseModel):
    message: str


class TestCallIn(BaseModel):
    number: str

@api_router.post("/agents/{agent_id}/test")
async def test_agent_simulator(agent_id: str, payload: TestMsgIn, user: dict = Depends(get_current_user)):
    """Text-based simulator that drives the agent's real LLM (Groq) with its persona + KB."""
    agent_doc = db.collection("agents").document(agent_id).get()
    if not agent_doc.exists or agent_doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=404, detail="Agent not found")
    agent = agent_doc.to_dict()

    # Build system prompt with persona + KB context
    system_prompt = agent.get("persona") or "You are a helpful AI voice assistant."
    kb_ids = agent.get("knowledge_base_ids") or []
    if kb_ids:
        kb_context = _fetch_kb_context(user["id"], kb_ids, payload.message)
        if kb_context:
            system_prompt += f"\n\nUse the following knowledge base context to answer when relevant:\n{kb_context}"

    started = datetime.now(timezone.utc)
    reply = await _llm_chat(
        provider=agent.get("llm_provider", "Groq"),
        model=agent.get("llm_model") or "llama-3.3-70b-versatile",
        system=system_prompt,
        user_msg=payload.message,
        language=agent.get("language", "Hinglish"),
    )
    latency = int((datetime.now(timezone.utc) - started).total_seconds() * 1000)
    return {"reply": reply, "latency_ms": latency}


_recent_dials: Dict[str, tuple] = {}   # key -> (call_id, expires_at)


@api_router.post("/agents/{agent_id}/test-call")
async def test_agent_call(agent_id: str, payload: TestCallIn, user: dict = Depends(get_current_user)):
    """Initiates an outbound SIP call via LiveKit to the provided number for testing the agent.

    Flow: Creates a LiveKit room → dispatches the AI agent → adds a SIP participant
    that dials the phone number. The agent and phone user talk through the LiveKit room.
    """
    agent_doc = db.collection("agents").document(agent_id).get()
    if not agent_doc.exists or agent_doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=404, detail="Agent not found")

    # Idempotency guard: React StrictMode / double-clicks / axios retries can fire
    # this twice within a second and place TWO phone calls. De-dupe the same
    # user+agent+number for a short window and return the in-flight call instead.
    dedupe_key = f"{user['id']}:{agent_id}:{(payload.number or '').strip()}"
    hit = _recent_dials.get(dedupe_key)
    if hit and hit[1] > monotonic():
        return {"ok": True, "call_id": hit[0], "deduped": True}

    # Resolve caller ID (from number)
    phone_docs = list(
        db.collection("phone_numbers")
        .where(filter=FieldFilter("assigned_agent_id", "==", agent_id))
        .limit(1)
        .stream()
    )
    from_number = None
    if phone_docs:
        from_number = phone_docs[0].to_dict().get("number")
    if not from_number:
        from_number = os.environ.get("VOBIZ_FROM_NUMBER", "").strip()
    if not from_number:
        raise HTTPException(status_code=400, detail="No phone number assigned to agent and no VOBIZ_FROM_NUMBER configured.")

    call_id = str(uuid.uuid4())
    # Register the dedupe guard BEFORE dialing so a near-simultaneous duplicate
    # request returns this call_id instead of placing a second call.
    _recent_dials[dedupe_key] = (call_id, monotonic() + 8)
    # Opportunistic cleanup of expired entries.
    for k in [k for k, v in _recent_dials.items() if v[1] <= monotonic()]:
        _recent_dials.pop(k, None)
    call_data = {
        "id": call_id,
        "direction": "outbound",
        "owner_id": user["id"],
        "agent_id": agent_id,
        "agent_name": agent_doc.to_dict().get("name", "Agent"),
        "phone_number": payload.number,
        "contact_name": "Test User",
        "started_at": datetime.now(timezone.utc).isoformat(),
        "status": "queued",
        "sentiment": "neutral",
        "lead_tag": "none",
        "intent": "",
        "summary": "",
        "analysis": "",
        "transcript": [],
        "duration_seconds": 0,
        "provider_call_id": None,
    }
    db.collection("calls").document(call_id).set(call_data)

    res = await _create_sip_outbound_call(
        agent_id=agent_id,
        owner_id=user["id"],
        to_number=payload.number,
        from_number=from_number,
        call_id=call_id,
    )

    if not res.get("ok"):
        db.collection("calls").document(call_id).update({"status": "failed", "summary": res.get("error")})
        raise HTTPException(status_code=500, detail=res.get("error", "SIP dial failed"))

    db.collection("calls").document(call_id).update({
        "status": "ringing",
        "room_name": res.get("room_name"),
    })
    _cache_invalidate_user(user["id"], "calls")
    return {"ok": True, "call_id": call_id}


async def _llm_chat(provider: str, model: str, system: str, user_msg: str, language: str) -> str:
    """Call the configured LLM provider. Falls back gracefully on errors."""
    lang_hint = {
        "Hindi": "Reply in natural Hindi (Devanagari or romanised).",
        "Hinglish": "Reply in natural Hinglish (mix Hindi+English casually).",
        "English": "Reply in clear, friendly English.",
    }.get(language, "")
    full_system = f"{system}\n\n{lang_hint}".strip()

    provider = (provider or "Groq").lower()
    try:
        if provider == "groq" or not provider:
            key = os.environ.get("GROQ_API_KEY")
            if not key:
                return "(LLM not configured — set GROQ_API_KEY)"
            if not model or "llama-3.3-70b" in model:
                model = "llama-3.3-70b-versatile"
            elif "llama-3.1-8b" in model:
                model = "llama-3.1-8b-instant"
            async with httpx.AsyncClient(timeout=30.0) as client:
                r = await client.post(
                    "https://api.groq.com/openai/v1/chat/completions",
                    headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
                    json={
                        "model": model,
                        "messages": [
                            {"role": "system", "content": full_system},
                            {"role": "user", "content": user_msg},
                        ],
                        "temperature": 0.6,
                        "max_tokens": 300,
                    },
                )
                r.raise_for_status()
                return r.json()["choices"][0]["message"]["content"].strip()
        if provider == "gemini":
            key = os.environ.get("GEMINI_API_KEY")
            if not key:
                return "(LLM not configured — set GEMINI_API_KEY)"
            model_id = model or "gemini-2.5-flash"
            async with httpx.AsyncClient(timeout=30.0) as client:
                r = await client.post(
                    f"https://generativelanguage.googleapis.com/v1beta/models/{model_id}:generateContent?key={key}",
                    json={
                        "system_instruction": {"parts": [{"text": full_system}]},
                        "contents": [{"parts": [{"text": user_msg}]}],
                    },
                )
                r.raise_for_status()
                cand = r.json().get("candidates", [])
                return cand[0]["content"]["parts"][0]["text"].strip() if cand else ""
        if provider in ("nvidia nim", "nvidia"):
            key = os.environ.get("NVIDIA_API_KEY")
            if not key:
                return "(LLM not configured — set NVIDIA_API_KEY)"
            async with httpx.AsyncClient(timeout=30.0) as client:
                r = await client.post(
                    "https://integrate.api.nvidia.com/v1/chat/completions",
                    headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
                    json={
                        "model": model or "meta/llama-3.1-70b-instruct",
                        "messages": [
                            {"role": "system", "content": full_system},
                            {"role": "user", "content": user_msg},
                        ],
                        "max_tokens": 300,
                    },
                )
                r.raise_for_status()
                return r.json()["choices"][0]["message"]["content"].strip()
        if provider in ("azure openai", "azure"):
            endpoint = os.environ.get("AZURE_OPENAI_ENDPOINT", "").rstrip("/")
            key = os.environ.get("AZURE_OPENAI_API_KEY", "")
            deployment = model or os.environ.get("AZURE_OPENAI_DEPLOYMENT", "gpt-4o")
            api_version = os.environ.get("AZURE_OPENAI_API_VERSION", "2024-12-01-preview")
            if not (endpoint and key):
                return "(LLM not configured — set AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_API_KEY)"
            async with httpx.AsyncClient(timeout=30.0) as client:
                r = await client.post(
                    f"{endpoint}/openai/deployments/{deployment}/chat/completions?api-version={api_version}",
                    headers={"api-key": key, "Content-Type": "application/json"},
                    json={
                        "messages": [
                            {"role": "system", "content": full_system},
                            {"role": "user", "content": user_msg},
                        ],
                        "temperature": 0.6,
                        "max_tokens": 300,
                    },
                )
                r.raise_for_status()
                return r.json()["choices"][0]["message"]["content"].strip()
    except httpx.HTTPStatusError as e:
        logger.error(f"LLM HTTP error: {e.response.status_code} {e.response.text[:200]}")
    except Exception as e:
        logger.error(f"LLM error: {type(e).__name__}: {e}")
    return "Sorry, I'm having trouble reaching the model right now. Please try again."


def _get_embedding(text: str, input_type: str = "passage") -> List[float]:
    """Generate embedding vector using NVIDIA NIM."""
    nvidia_key = os.getenv("NVIDIA_API_KEY")
    if not nvidia_key:
        logger.warning("NVIDIA_API_KEY not set. Cannot generate embeddings.")
        return []
    
    try:
        client = OpenAI(
            api_key=nvidia_key,
            base_url="https://integrate.api.nvidia.com/v1"
        )
        response = client.embeddings.create(
            input=[text],
            model="nvidia/llama-nemotron-embed-1b-v2",
            encoding_format="float",
            extra_body={"input_type": input_type, "truncate": "NONE"}
        )
        return response.data[0].embedding
    except Exception as e:
        logger.error(f"Failed to generate embedding: {e}")
        return []

def _cosine_similarity(vec1: List[float], vec2: List[float]) -> float:
    if not vec1 or not vec2:
        return 0.0
    v1 = np.array(vec1)
    v2 = np.array(vec2)
    norm1 = np.linalg.norm(v1)
    norm2 = np.linalg.norm(v2)
    if norm1 == 0 or norm2 == 0:
        return 0.0
    return float(np.dot(v1, v2) / (norm1 * norm2))


def _fetch_kb_context(owner_id: str, kb_ids: List[str], query: str, max_chunks: int = 4) -> str:
    """Vector search retrieval using NVIDIA NIM embeddings."""
    if not query.strip() or not kb_ids:
        return ""
        
    query_vector = _get_embedding(query, input_type="query")
    if not query_vector:
        # Fallback to keyword search if embedding fails
        query_terms = {w.lower() for w in re.findall(r"\w+", query) if len(w) > 2}
        if not query_terms:
            return ""
        chunks: List[Dict[str, Any]] = []
        for kid in kb_ids:
            kdoc = db.collection("knowledge").document(kid).get()
            if not kdoc.exists or kdoc.to_dict().get("owner_id") != owner_id:
                continue
            for sub in db.collection("knowledge").document(kid).collection("chunks").stream():
                chunks.append(sub.to_dict())
        scored = []
        for c in chunks:
            text = (c.get("text") or "").lower()
            score = sum(1 for t in query_terms if t in text)
            if score > 0:
                scored.append((score, c.get("text", "")))
        scored.sort(key=lambda x: x[0], reverse=True)
        top = [s[1] for s in scored[:max_chunks]]
        return "\n\n".join(top)

    chunks: List[Dict[str, Any]] = []
    for kid in kb_ids:
        kdoc = db.collection("knowledge").document(kid).get()
        if not kdoc.exists or kdoc.to_dict().get("owner_id") != owner_id:
            continue
        for sub in db.collection("knowledge").document(kid).collection("chunks").stream():
            chunks.append(sub.to_dict())
            
    scored = []
    for c in chunks:
        text = c.get("text", "")
        vec = c.get("vector")
        if vec and isinstance(vec, list):
            score = _cosine_similarity(query_vector, vec)
        else:
            # Fallback score if chunk has no embedding
            score = 0.0 
        scored.append((score, text))
        
    scored.sort(key=lambda x: x[0], reverse=True)
    top = [s[1] for s in scored[:max_chunks] if s[0] > 0.1]  # Threshold 0.1
    return "\n\n".join(top)


# ----- Campaigns -----
@api_router.get("/campaigns", response_model=List[Campaign])
def list_campaigns(user: dict = Depends(get_current_user)):
    cached = _cache_get(f"campaigns:{user['id']}")
    if cached is not None:
        return cached
    docs = db.collection("campaigns").where(filter=FieldFilter("owner_id", "==", user["id"])).stream()
    return _cache_set(f"campaigns:{user['id']}", [doc.to_dict() for doc in docs], ttl=15)


@api_router.post("/campaigns", response_model=Campaign)
def create_campaign(payload: CampaignIn, user: dict = Depends(get_current_user)):
    if not payload.name.strip():
        raise HTTPException(status_code=400, detail="Campaign name is required")
    cid = str(uuid.uuid4())
    total = len(payload.contacts)
    doc_data = {
        "id": cid,
        **payload.model_dump(),
        "owner_id": user["id"],
        "status": "draft",
        "created_at": now_iso(),
        "progress": {"total": total, "dialed": 0, "connected": 0, "completed": 0, "failed": 0, "leads": 0},
    }
    db.collection("campaigns").document(cid).set(doc_data)
    _cache_invalidate_user(user["id"], "campaigns")
    return {k: v for k, v in doc_data.items() if k != "owner_id"}


@api_router.post("/campaigns/upload-csv")
async def campaign_upload_csv(file: UploadFile = File(...), user: dict = Depends(get_current_user)):
    if not file.filename.lower().endswith(".csv"):
        raise HTTPException(status_code=400, detail="Only .csv files are supported")
    content = (await file.read()).decode("utf-8", errors="ignore")
    reader = csv.DictReader(io.StringIO(content))
    contacts = []
    if reader.fieldnames and any(("phone" in (h or "").lower()) for h in reader.fieldnames):
        for row in reader:
            phone = row.get("phone") or row.get("Phone") or row.get("number") or row.get("Number") or ""
            name = row.get("name") or row.get("Name") or "Unknown"
            if phone.strip():
                contacts.append({"name": name.strip(), "phone": phone.strip()})
    else:
        # Fall back to positional parsing (name, phone)
        for line in content.splitlines():
            parts = [p.strip() for p in line.split(",")]
            if len(parts) >= 2 and parts[1]:
                contacts.append({"name": parts[0] or "Unknown", "phone": parts[1]})
    return {"contacts": contacts, "count": len(contacts)}


@api_router.post("/campaigns/{campaign_id}/start", response_model=Campaign)
def start_campaign(campaign_id: str, background: BackgroundTasks, user: dict = Depends(get_current_user)):
    camp_ref = db.collection("campaigns").document(campaign_id)
    camp_doc = camp_ref.get()
    if not camp_doc.exists or camp_doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=404, detail="Campaign not found")
    data = camp_doc.to_dict()
    if data.get("status") == "running":
        raise HTTPException(status_code=400, detail="Campaign already running")

    camp_ref.update({"status": "running"})
    _cache_invalidate_user(user["id"], "campaigns")
    background.add_task(_run_campaign_loop, campaign_id, user["id"])
    return camp_ref.get().to_dict()


@api_router.post("/campaigns/{campaign_id}/pause", response_model=Campaign)
def pause_campaign(campaign_id: str, user: dict = Depends(get_current_user)):
    camp_ref = db.collection("campaigns").document(campaign_id)
    camp_doc = camp_ref.get()
    if not camp_doc.exists or camp_doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=404, detail="Campaign not found")
    camp_ref.update({"status": "paused"})
    _cache_invalidate_user(user["id"], "campaigns")
    return camp_ref.get().to_dict()


@api_router.delete("/campaigns/{campaign_id}")
def delete_campaign(campaign_id: str, user: dict = Depends(get_current_user)):
    camp_ref = db.collection("campaigns").document(campaign_id)
    camp_doc = camp_ref.get()
    if not camp_doc.exists or camp_doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=404, detail="Campaign not found")
    camp_ref.delete()
    _cache_invalidate_user(user["id"], "campaigns")
    return {"ok": True}


async def _run_campaign_loop(campaign_id: str, owner_id: str):
    """Background dialing loop: iterates contacts, dials via Vobiz, creates call records,
    respects pacing_per_minute, stops when campaign is paused/deleted."""
    try:
        camp_ref = db.collection("campaigns").document(campaign_id)
        snap = camp_ref.get()
        if not snap.exists:
            return
        camp = snap.to_dict()
        agent_doc = db.collection("agents").document(camp["agent_id"]).get()
        agent = agent_doc.to_dict() if agent_doc.exists else {"name": "Agent", "id": camp["agent_id"]}
        contacts = camp.get("contacts", [])
        pacing = max(1, int(camp.get("pacing_per_minute", 10)))
        interval = 60.0 / pacing

        progress = dict(camp.get("progress") or {})
        for contact in contacts:
            # Re-check status each iteration to honour pause/delete
            cur = camp_ref.get()
            if not cur.exists:
                return
            if cur.to_dict().get("status") != "running":
                logger.info(f"Campaign {campaign_id} stopped ({cur.to_dict().get('status')})")
                return

            phone = (contact.get("phone") or "").strip()
            name = (contact.get("name") or "Unknown").strip()
            if not phone:
                continue

            # Create call record BEFORE dialling so the webhook can resolve it
            call_id = str(uuid.uuid4())
            db.collection("calls").document(call_id).set({
                "id": call_id,
                "agent_id": agent.get("id"),
                "agent_name": agent.get("name", "Agent"),
                "direction": "outbound",
                "phone_number": phone,
                "contact_name": name,
                "duration_seconds": 0,
                "status": "queued",
                "sentiment": "neutral",
                "lead_tag": "none",
                "intent": "",
                "summary": "",
                "transcript": [],
                "started_at": now_iso(),
                "campaign_id": campaign_id,
                "owner_id": owner_id,
                "provider_call_id": None,
                "recording_url": None,
            })
            _cache_invalidate_user(owner_id, "calls", "analytics", "leaderboard")

            from_number = agent.get("phone_number") or os.environ.get("VOBIZ_FROM_NUMBER", "")

            result = await _create_sip_outbound_call(
                agent_id=agent.get("id", ""),
                owner_id=owner_id,
                to_number=phone,
                from_number=from_number,
                call_id=call_id,
            )

            if result.get("ok"):
                db.collection("calls").document(call_id).update({
                    "status": "dialing",
                    "room_name": result.get("room_name"),
                })
                progress["dialed"] = progress.get("dialed", 0) + 1
            else:
                db.collection("calls").document(call_id).update({
                    "status": "failed",
                    "summary": result.get("error", "Dial failed"),
                })
                progress["failed"] = progress.get("failed", 0) + 1
            _cache_invalidate_user(owner_id, "calls", "analytics", "leaderboard")

            camp_ref.update({"progress": progress})
            _cache_invalidate_user(owner_id, "campaigns")
            await asyncio.sleep(interval)

        camp_ref.update({"status": "completed"})
        _cache_invalidate_user(owner_id, "campaigns")
    except Exception as e:
        logger.exception(f"Campaign loop error: {e}")
        try:
            db.collection("campaigns").document(campaign_id).update({"status": "failed"})
        except Exception:
            pass


async def _dial_via_vobiz(to_number: str, from_number: str, call_id: str) -> Dict[str, Any]:
    """Place an outbound call via Vobiz REST API.

    Env vars used:
      VOBIZ_AUTH_ID      — API auth user ID
      VOBIZ_AUTH_SECRET  — API auth secret
      VOBIZ_SIP_DOMAIN   — API host, e.g. api.vobiz.in
      VOBIZ_FROM_NUMBER  — Caller-ID / DID number (fallback)
      VOBIZ_CALLBACK_URL — Public base URL for status webhook (optional)
    Returns: {ok, call_uuid} or {ok: False, error}
    """
    auth_id = os.environ.get("VOBIZ_AUTH_ID", "").strip()
    auth_secret = os.environ.get("VOBIZ_AUTH_SECRET", "").strip()
    sip_domain = os.environ.get("VOBIZ_SIP_DOMAIN", "").strip().rstrip("/")
    callback_base = os.environ.get("VOBIZ_CALLBACK_URL", "").strip().rstrip("/")

    if not (auth_id and auth_secret and sip_domain):
        logger.warning("Vobiz credentials not configured (VOBIZ_AUTH_ID/SECRET/SIP_DOMAIN)")
        return {"ok": False, "error": "Vobiz not configured"}

    if not from_number:
        return {"ok": False, "error": "Caller number not set (add phone number to agent or set VOBIZ_FROM_NUMBER)"}

    # Vobiz click-to-call REST endpoint
    url = f"https://{sip_domain}/api/v1/call/outbound/"
    body: Dict[str, Any] = {
        "from": from_number,
        "to": to_number,
        "caller_id": from_number,
        "custom_data": call_id,  # echoed back in webhook so we can resolve the record
    }
    if callback_base:
        body["answer_url"] = f"{callback_base}/api/webhooks/vobiz"
        body["hangup_url"] = f"{callback_base}/api/webhooks/vobiz"

    try:
        async with httpx.AsyncClient(timeout=20.0) as client:
            r = await client.post(
                url,
                json=body,
                auth=(auth_id, auth_secret),
                headers={"Content-Type": "application/json"},
            )
            if r.status_code >= 400:
                logger.error(f"Vobiz dial {r.status_code}: {r.text[:300]}")
                return {"ok": False, "error": f"Vobiz API {r.status_code}: {r.text[:200]}"}
            data = r.json()
            # Vobiz returns call_uuid or api_id depending on API version
            call_uuid = (
                data.get("call_uuid")
                or data.get("api_id")
                or data.get("id")
                or data.get("uuid")
            )
            logger.info(f"Vobiz dial ok: to={to_number} uuid={call_uuid}")
            return {"ok": True, "call_uuid": call_uuid}
    except Exception as e:
        logger.error(f"Vobiz dial exception: {type(e).__name__}: {e}")
        return {"ok": False, "error": str(e)}


# ----- Vobiz Webhook -----
@api_router.post("/webhooks/vobiz")
async def vobiz_webhook(request: Request):
    """Receives call-status events from Vobiz.

    Vobiz sends JSON with these fields (varies by API version):
      call_uuid / api_id       — provider call identifier
      custom_data              — our call_id (set at dial time)
      event / status           — e.g. answered, completed, failed, no-answer, busy
      duration / bill_duration — call duration in seconds
      recording_url            — URL of the recording (if enabled)
      transcript               — transcript text (if enabled)
    """
    try:
        payload = await request.json()
    except Exception:
        form = await request.form()
        payload = dict(form)

    logger.info(f"Vobiz webhook: {json.dumps(payload, default=str)[:500]}")

    # Resolve provider call ID (Vobiz field names differ by API version)
    provider_call_id = (
        payload.get("call_uuid")
        or payload.get("api_id")
        or payload.get("uuid")
        or payload.get("call_id")
    )
    # Our internal call_id echoed back via custom_data
    our_call_id = payload.get("custom_data") or payload.get("custom_field")

    if not (provider_call_id or our_call_id):
        return {"status": "ignored", "reason": "no identifiable call_id"}

    # Locate the call document
    doc_ref = None
    if our_call_id:
        cand = db.collection("calls").document(our_call_id).get()
        if cand.exists:
            doc_ref = cand.reference
    if not doc_ref and provider_call_id:
        found = list(
            db.collection("calls")
            .where(filter=FieldFilter("provider_call_id", "==", provider_call_id))
            .limit(1)
            .stream()
        )
        if found:
            doc_ref = found[0].reference
    if not doc_ref:
        logger.warning(f"Vobiz webhook — no matching call: uuid={provider_call_id} custom={our_call_id}")
        return {"status": "no_match"}

    update: Dict[str, Any] = {"updated_at": now_iso()}

    # Backfill provider_call_id if we only had our_call_id
    if provider_call_id:
        update["provider_call_id"] = provider_call_id

    # Status mapping
    raw_status = (
        payload.get("event")
        or payload.get("status")
        or payload.get("call_status")
        or ""
    ).lower()
    STATUS_MAP = {
        "completed": "completed",
        "answered":  "completed",
        "hangup":    "completed",
        "no-answer": "missed",
        "noanswer":  "missed",
        "missed":    "missed",
        "busy":      "failed",
        "failed":    "failed",
        "rejected":  "failed",
        "canceled":  "failed",
        "cancelled":  "failed",
    }
    if raw_status:
        update["status"] = STATUS_MAP.get(raw_status, raw_status)

    # Duration (Vobiz may send bill_duration, duration, or conversation_duration)
    for dur_key in ("bill_duration", "duration", "conversation_duration", "call_duration"):
        raw_dur = payload.get(dur_key)
        if raw_dur is not None:
            try:
                update["duration_seconds"] = int(float(raw_dur))
            except Exception:
                pass
            break

    # Recording
    for rec_key in ("recording_url", "RecordingUrl", "record_url"):
        if payload.get(rec_key):
            update["recording_url"] = payload[rec_key]
            break

    # Transcript (plain text — Vobiz may include after STT processing)
    for tr_key in ("transcript", "transcription", "transcription_text"):
        if payload.get(tr_key):
            update["transcript_text"] = payload[tr_key]
            break

    existing_call = doc_ref.get().to_dict() or {}
    owner_id = existing_call.get("owner_id")
    doc_ref.update(update)

    # Run AI analysis on completed calls with transcripts
    is_final = update.get("status") in ("completed", "missed", "failed")
    transcript_text = update.get("transcript_text")

    # Fallback: if transcript_text is empty, but existing_call has a transcript (list of turns)
    if is_final and not transcript_text and existing_call.get("transcript"):
        try:
            turns = existing_call["transcript"]
            if isinstance(turns, list):
                transcript_text = "\n".join(f"{t.get('speaker', 'unknown')}: {t.get('text', '')}" for t in turns if isinstance(t, dict))
        except Exception as te:
            logger.warning(f"Failed to rebuild transcript text from log: {te}")

    if is_final and transcript_text and transcript_text.strip():
        try:
            insights = await _analyze_transcript(transcript_text)
            if insights:
                lead = insights.get("lead") or {}
                known_phone = (existing_call.get("phone_number") or "").strip()
                lead_phone = (known_phone or str(lead.get("phone") or "")).strip().lstrip("+")
                
                analysis_updates = {
                    "sentiment": insights.get("sentiment", "neutral"),
                    "lead_tag": insights.get("lead_tag", "none"),
                    "intent": insights.get("intent", ""),
                    "summary": insights.get("summary", ""),
                    "analysis": insights.get("analysis", ""),
                    "outcome": insights.get("outcome", ""),
                    "topics": insights.get("topics", []),
                    "lead_name": lead.get("name", ""),
                    "lead_email": lead.get("email", ""),
                    "lead_company": lead.get("company", ""),
                    "lead_phone": lead_phone,
                    "lead_score": lead.get("score", 0),
                    "requested_materials": insights.get("requested_materials", []),
                    "action_items": insights.get("action_items", []),
                }
                update.update(analysis_updates)
                doc_ref.update(update)
                
                # Also upsert the lead
                agent_id = existing_call.get("agent_id") or ""
                agent_name = existing_call.get("agent_name") or "Agent"
                log_data = {
                    "id": existing_call.get("id") or our_call_id or doc_ref.id,
                    "lead_phone": lead_phone,
                    "summary": insights.get("summary", ""),
                    "sentiment": insights.get("sentiment", "neutral"),
                }
                
                _upsert_lead(
                    owner_id=owner_id,
                    agent_id=agent_id,
                    agent_name=agent_name,
                    log_data=log_data,
                    insights=insights,
                    source="voice_sip"
                )
        except Exception as e:
            logger.error(f"Transcript analysis failed: {e}")

    if owner_id:
        _cache_invalidate_user(owner_id, "calls", "analytics", "leaderboard")
    return {"status": "ok"}


async def _analyze_transcript(text: str) -> Dict[str, Any]:
    """Extract structured post-call intelligence: sentiment, lead fields,
    follow-up, requested materials, action items.

    Uses Groq FIRST (fast, reliable); NVIDIA NIM (z-ai/glm-5.2, thinking OFF) as
    fallback because integrate.api.nvidia.com throws intermittent 503s under load.
    Both use JSON-mode + a low temperature."""
    if not text.strip():
        return {}
    system = (
        "You are an expert call analyst. Analyze the transcript and return a STRICT JSON object with fields:\n"
        '- sentiment: "positive" | "neutral" | "negative" (the CUSTOMER\'s attitude; when unsure use "neutral").\n'
        '- lead_tag: "hot" | "warm" | "cold" | "none".\n'
        "- intent: short string, the customer's primary goal.\n"
        "- outcome: short string, the final resolution.\n"
        "- summary: a 2-3 sentence summary.\n"
        "- analysis: a detailed paragraph (mood, pain points, next steps).\n"
        "- topics: array of up to 5 short topic strings.\n"
        '- lead: {"name","email","company","phone","score" (0-10)} — "" for anything not mentioned; '
        "never invent email/phone digits (take them verbatim from what the caller said).\n"
        '- follow_up: {"datetime": ISO8601 or "", "topic": ""}.\n'
        '- requested_materials: array (e.g. "case_study","pricing","brochure","demo") or [].\n'
        "- action_items: array of strings or [].\n"
    )
    messages = [{"role": "system", "content": system},
                {"role": "user", "content": f"Transcript:\n{text[:10000]}"}]

    # NVIDIA NIM FIRST per user config; Groq as reliable fallback if NIM is unavailable/503s.
    nvidia_key = os.environ.get("NVIDIA_API_KEY")
    groq_key = os.environ.get("GROQ_API_KEY")
    nim_model = os.environ.get("NIM_ANALYTICS_MODEL", "z-ai/glm-5.2")
    groq_model = os.environ.get("ANALYTICS_MODEL", "llama-3.3-70b-versatile")
    providers = []
    if nvidia_key:
        providers.append(("NVIDIA", "https://integrate.api.nvidia.com/v1", nvidia_key, nim_model, True))
    if groq_key:
        providers.append(("Groq", "https://api.groq.com/openai/v1", groq_key, groq_model, False))

    for label, base_url, api_key, model, is_nvidia in providers:
        try:
            body = {"model": model, "messages": messages,
                    "response_format": {"type": "json_object"}, "temperature": 0.1,
                    "max_tokens": 1024}
            # Reasoning models on NIM (Nemotron/GLM): turn thinking OFF so we get
            # the final JSON immediately instead of a slow reasoning trace.
            if is_nvidia and model.startswith(("nvidia/nemotron", "z-ai/glm")):
                body["chat_template_kwargs"] = {"enable_thinking": False}
            async with httpx.AsyncClient(timeout=30.0) as client:
                r = await client.post(
                    f"{base_url}/chat/completions",
                    headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
                    json=body)
                r.raise_for_status()
                content = r.json()["choices"][0]["message"]["content"]
            content = re.sub(r"<think>.*?</think>", "", content, flags=re.DOTALL).strip()
            m = re.search(r"\{[\s\S]*\}", content)
            parsed = json.loads(m.group(0) if m else content)
            logger.info(f"[Analytics] webhook transcript analyzed via {label}")
            lead = parsed.get("lead") if isinstance(parsed.get("lead"), dict) else {}
            score_raw = str(lead.get("score") or "0")
            return {
                "sentiment": (parsed.get("sentiment") or "neutral").lower(),
                "lead_tag": (parsed.get("lead_tag") or "none").lower(),
                "intent": parsed.get("intent", ""),
                "outcome": parsed.get("outcome", ""),
                "summary": parsed.get("summary", ""),
                "analysis": parsed.get("analysis", ""),
                "topics": (parsed.get("topics") or [])[:5],
                "lead": {
                    "name": str(lead.get("name") or "")[:120],
                    "email": str(lead.get("email") or "")[:200],
                    "company": str(lead.get("company") or "")[:160],
                    "phone": str(lead.get("phone") or "")[:40],
                    "score": int(score_raw) if score_raw.isdigit() else 0,
                },
                "follow_up": parsed.get("follow_up") if isinstance(parsed.get("follow_up"), dict) else {},
                "requested_materials": parsed.get("requested_materials") or [],
                "action_items": parsed.get("action_items") or [],
            }
        except Exception as e:
            logger.warning(f"[Analytics] {label} failed ({e}) — trying next")
    logger.error("Analyze transcript: all providers errored")
    return {}


def _upsert_lead(owner_id: str, agent_id: str, agent_name: str, log_data: dict,
                 insights: dict, source: str):
    """Create or refresh a lead (deduped by phone) so the Leads view is populated."""
    if not db or not owner_id:
        return
    try:
        lead = insights.get("lead") or {}
        phone = (log_data.get("lead_phone") or lead.get("phone") or "").strip().lstrip("+")
        score = int(lead.get("score") or 0)
        status = "hot" if score >= 7 else ("warm" if score >= 4 else "new")
        payload = {
            "owner_id": owner_id, "agent_id": agent_id, "agent_name": agent_name,
            "name": lead.get("name", ""), "email": lead.get("email", ""),
            "company": lead.get("company", ""), "phone": phone,
            "intent": insights.get("intent", ""), "message": log_data.get("summary", ""),
            "source": source, "status": status, "sentiment": log_data.get("sentiment", "neutral"),
            "score": score, "last_call_id": log_data.get("id", ""),
            "updated_at": firestore.SERVER_TIMESTAMP,
        }
        existing = None
        if phone:
            try:
                existing = list(db.collection("leads")
                                .where(filter=FieldFilter("owner_id", "==", owner_id))
                                .where(filter=FieldFilter("phone", "==", phone)).limit(1).stream())
            except Exception as qe:
                logger.warning(f"[Lead] dedup query failed ({qe}); creating new")
                existing = None
        if existing:
            db.collection("leads").document(existing[0].id).set(payload, merge=True)
            logger.info(f"[Lead] updated for {phone}")
        elif phone or lead.get("name") or lead.get("email"):
            lid = "lead_" + uuid.uuid4().hex[:20]
            payload["id"] = lid
            payload["created_at"] = firestore.SERVER_TIMESTAMP
            db.collection("leads").document(lid).set(payload)
            logger.info(f"[Lead] captured {lid} for {phone or '(no phone)'}")
    except Exception as e:
        logger.warning(f"[Lead] upsert failed: {e}")


def _save_sheet_crm(owner_id: str, agent_data: dict, log_data: dict, insights: dict) -> bool:
    """Write/refresh the caller's row in the agent's Google Sheet CRM, keyed by
    phone (the caller's unique id). Returns True if written."""
    try:
        import integrations as _integrations
    except ImportError:
        _integrations = None
    if _integrations is None or not owner_id:
        return False
    crm_sheet_id = (agent_data or {}).get("crm_spreadsheet_id") or ""
    phone = (log_data.get("lead_phone") or log_data.get("phone_number") or "").strip()
    if not (crm_sheet_id and phone):
        return False
    if not _integrations.is_connected(owner_id, "google_sheets"):
        return False
    if _integrations.is_connected_any(owner_id, ["hubspot", "zoho"]):
        return False  # a real CRM is connected → use that instead
    phone = phone.lstrip("+")
    try:
        # Bump call count if the caller already exists.
        prev = _integrations.execute_tool(owner_id, "read_sheet_rows", {
            "spreadsheet_id": crm_sheet_id, "sheet_name": "CRM",
            "search_column": "A", "search_value": phone})
        count = 1
        rows = (prev or {}).get("rows") or []
        old = rows[0] if rows else []
        if rows:
            # This async path ENRICHES the row the agent already saved for this
            # same call — do NOT bump the call count again (double-counting).
            try:
                count = int(old[7]) if len(old) > 7 and str(old[7]).isdigit() else 1
            except Exception:
                count = 1

        def _keep(new_val, idx):
            # Never clobber good data with blanks — preserve the previous value.
            new_val = (new_val or "").strip() if isinstance(new_val, str) else (new_val or "")
            if new_val:
                return new_val
            return old[idx] if len(old) > idx and old[idx] else ""

        # Columns: Phone|Email|Name|Company|Intent|Last Call Summary|Last Call Date|Call Count|Status
        row = [phone,
               _keep(log_data.get("lead_email", ""), 1),
               _keep(log_data.get("lead_name", ""), 2),
               _keep(log_data.get("lead_company", ""), 3),
               _keep((insights.get("intent") or "")[:200], 4),
               _keep((log_data.get("summary") or "")[:500], 5),
               datetime.now().isoformat()[:19],
               str(count),
               _keep(log_data.get("lead_tag", ""), 8) or "active"]
        res = _integrations.execute_tool(owner_id, "update_sheet_row", {
            "spreadsheet_id": crm_sheet_id, "key_column": "A", "key_value": phone,
            "values": row, "sheet_name": "CRM"})
        ok = bool((res or {}).get("ok"))
        logger.info(f"[SheetCRM] {'saved' if ok else 'FAILED to save'} caller {phone} "
                    f"(call #{count})")
        return ok
    except Exception as e:
        logger.warning(f"[SheetCRM] save failed: {e}")
        return False


async def _process_async_analysis(call_id: str, call_data: dict):
    logger.info(f"Starting async post-call analysis for call {call_id}")
    
    # 1. Rebuild the transcript from log
    turns = call_data.get("transcript") or []
    transcript_text = ""
    if isinstance(turns, list) and turns:
        transcript_text = "\n".join(f"{t.get('speaker', 'unknown')}: {t.get('text', '')}" for t in turns if isinstance(t, dict))
    elif call_data.get("transcript_text"):
        transcript_text = call_data["transcript_text"]
        
    if not transcript_text.strip():
        logger.warning(f"No transcript found for call {call_id}, skipping analysis")
        return
        
    # 2. Run _analyze_transcript with robust retries (up to 5 times)
    insights = {}
    for attempt in range(1, 6):
        try:
            # Persistent background task has no tight limits, use 25s timeout
            insights = await asyncio.wait_for(_analyze_transcript(transcript_text), timeout=25.0)
            if insights:
                break
        except Exception as e:
            logger.warning(f"Analysis attempt {attempt} failed: {e}")
            await asyncio.sleep(3 * attempt)
            
    if not insights:
        logger.error(f"Analysis failed for call {call_id} after 5 attempts")
        return
        
    # 3. Save insights to Firestore
    lead = insights.get("lead") or {}
    known_phone = (call_data.get("phone_number") or "").strip()
    lead_phone = (known_phone or str(lead.get("phone") or "")).strip().lstrip("+")
    
    analysis_updates = {
        "sentiment": insights.get("sentiment", "neutral"),
        "lead_tag": insights.get("lead_tag", "none"),
        "intent": insights.get("intent", ""),
        "summary": insights.get("summary", ""),
        "analysis": insights.get("analysis", ""),
        "outcome": insights.get("outcome", ""),
        "topics": insights.get("topics", []),
        "lead_name": lead.get("name", ""),
        "lead_email": lead.get("email", ""),
        "lead_company": lead.get("company", ""),
        "lead_phone": lead_phone,
        "lead_score": lead.get("score", 0),
        "requested_materials": insights.get("requested_materials", []),
        "action_items": insights.get("action_items", []),
    }
    
    try:
        db.collection("calls").document(call_id).update(analysis_updates)
        logger.info(f"Firestore call log {call_id} updated with insights")
    except Exception as fe:
        logger.error(f"Failed to update call log {call_id} with insights: {fe}")
    
    # 4. Upsert lead
    owner_id = call_data.get("owner_id")
    agent_id = call_data.get("agent_id")
    agent_name = call_data.get("agent_name") or "Agent"
    log_data = {
        "id": call_id,
        "lead_phone": lead_phone,
        "summary": insights.get("summary", ""),
        "sentiment": insights.get("sentiment", "neutral"),
    }
    
    try:
        _upsert_lead(
            owner_id=owner_id,
            agent_id=agent_id,
            agent_name=agent_name,
            log_data=log_data,
            insights=insights,
            source=call_data.get("channel", "voice_web")
        )
    except Exception as le:
        logger.warning(f"Async Lead upsert failed: {le}")
        
    # 5. Save to Google Sheets CRM (if agent data is present)
    agent_data = {}
    try:
        if agent_id:
            agent_doc = db.collection("agents").document(agent_id).get()
            if agent_doc.exists:
                agent_data = agent_doc.to_dict() or {}
                _save_sheet_crm(owner_id, agent_data, {**call_data, **log_data, **analysis_updates}, insights)
    except Exception as se:
        logger.warning(f"Async Sheet CRM save failed: {se}")

    # 6. POST-CALL AUTOMATION (email + calendar) — this is the ONLY reliable place
    # for it: the agent worker's local automation usually runs with EMPTY insights
    # (its analysis times out and defers here), so the follow-up email/invite never
    # fired. Here we have the full insights and no exit deadline.
    try:
        actions = await asyncio.to_thread(
            _post_call_automation, owner_id, agent_data, call_data, analysis_updates, insights)
        if actions:
            db.collection("calls").document(call_id).set({"actions": actions}, merge=True)
            logger.info(f"[PostCall] {len(actions)} action(s) recorded for {call_id}")
    except Exception as pe:
        logger.warning(f"[PostCall] async automation failed: {pe}")


def _post_call_automation(owner_id, agent_data, call_data, updates, insights):
    """Fire the follow-up calendar invite + email from the async analysis path.
    Calendar first (so the email can carry the meeting + Meet link); email carries
    ONLY the meeting details and/or requested materials — never a summary recap."""
    import integrations as _int
    actions = []
    if not owner_id:
        return actions
    agent_integrations = (agent_data or {}).get("integrations") or {}

    def _allows(key):
        return agent_integrations.get(key) is not False

    lead_name = (updates.get("lead_name") or "").strip()
    lead_email = (updates.get("lead_email") or "").strip()
    follow_up = insights.get("follow_up") or {}
    materials = [m for m in (insights.get("requested_materials") or []) if m]

    # --- Calendar invite (clash-free) ---
    meeting = None
    fu_dt = (follow_up.get("datetime") or "").strip()
    fu_topic = (follow_up.get("topic") or "Follow-up call").strip()
    action_items = insights.get("action_items") or updates.get("action_items") or []
    topics = insights.get("topics") or updates.get("topics") or []
    summary = updates.get("summary") or "It was great speaking with you."

    if not fu_dt or len(fu_dt) <= 8:
        combined_text = " ".join([
            str(updates.get("analysis") or ""),
            str(summary),
            str(updates.get("intent") or ""),
            " ".join(str(x) for x in action_items),
            " ".join(str(x) for x in topics)
        ]).lower()
        if any(w in combined_text for w in ("tomorrow", "calendar", "meeting", "reschedule", "demo", "invite", "follow-up call", "schedule")):
            next_day = datetime.now(timezone.utc) + timedelta(days=1)
            fu_dt = next_day.replace(hour=5, minute=30, second=0, microsecond=0).isoformat()
            if not fu_topic or fu_topic == "Follow-up call":
                fu_topic = "Follow-up Meeting & Discovery"

    if fu_dt and len(fu_dt) > 8 and _allows("google_calendar") and _int.is_connected(owner_id, "google_calendar"):
        try:
            start = datetime.fromisoformat(fu_dt.replace("Z", "+00:00"))
            start, end, shifted = _int._find_clash_free_slot(owner_id, start, 30)
            invite = {"title": fu_topic, "start": start.isoformat(), "end": end.isoformat(),
                      "description": "Follow-up scheduled from your call.",
                      "tz": "Asia/Kolkata", "create_meet_link": True}
            if lead_email and "@" in lead_email:
                invite["attendees"] = [lead_email]  # Google emails them the invite too
            res = _int.execute_tool(owner_id, "create_calendar_event", invite)
            if (res or {}).get("ok"):
                meeting = {"when": start.strftime("%A, %d %b %Y at %I:%M %p"),
                           "meet_link": res.get("meet_link") or res.get("html_link") or "",
                           "topic": fu_topic}
            detail = f"{fu_topic} @ {start.strftime('%d %b %H:%M')}" + (" (auto-shifted)" if shifted else "")
            actions.append({"type": "calendar", "status": "ok" if (res or {}).get("ok") else "failed",
                            "label": "Scheduled follow-up meeting", "detail": detail})
            logger.info(f"[PostCall] calendar: {detail} ok={bool((res or {}).get('ok'))}")
        except Exception as e:
            logger.warning(f"[PostCall] calendar failed: {e}")
    elif not (fu_dt and len(fu_dt) > 8):
        logger.info("[PostCall] calendar skipped — no follow-up datetime found in the call "
                    "(the caller must agree a concrete day/time for an invite)")

    # --- Email: when there's meeting invite, materials, or AI analysis next steps
    if not materials and topics:
        materials = [m for m in topics if m and str(m).lower() not in ("none", "general", "other")]
    has_actionable = bool(meeting or materials or action_items or updates.get("analysis"))
    if lead_email and "@" in lead_email and _allows("email") and has_actionable:
        try:
            first = lead_name.split(" ")[0] if lead_name else "there"
            blocks = []
            if meeting:
                link = meeting.get("meet_link") or ""
                link_html = (f"<p style='margin:6px 0 0;'><a href='{link}' "
                             f"style='color:#e0533d;font-weight:600;'>Join the meeting →</a></p>") if link else ""
                blocks.append(
                    f"<div style='margin:0 0 18px;padding:14px 16px;background:#fff5f3;"
                    f"border:1px solid #f7c9bf;border-radius:10px;'>"
                    f"<p style='margin:0;font-size:13px;text-transform:uppercase;letter-spacing:.05em;"
                    f"color:#c2410c;font-weight:700;'>Your meeting is booked</p>"
                    f"<p style='margin:6px 0 0;font-size:15px;'><strong>{meeting.get('topic', 'Follow-up')}</strong></p>"
                    f"<p style='margin:2px 0 0;font-size:15px;color:#333;'>{meeting['when']}</p>{link_html}</div>")
            if materials:
                li = "".join(f"<li style='margin:4px 0;'>{m.replace('_', ' ').title()}</li>" for m in materials[:6])
                blocks.append(
                    f"<p style='font-size:15px;font-weight:600;margin:8px 0;'>The information you asked for:</p>"
                    f"<ul style='margin:0 0 8px;padding-left:20px;color:#333;'>{li}</ul>")
            if action_items or summary:
                ai_li = "".join(f"<li style='margin:4px 0;'>{a}</li>" for a in action_items[:5])
                blocks.append(
                    f"<div style='margin:14px 0;padding:14px 16px;background:#f8fafc;border:1px solid #e2e8f0;border-radius:10px;'>"
                    f"<p style='margin:0 0 6px;font-size:13px;text-transform:uppercase;letter-spacing:.05em;color:#475569;font-weight:700;'>Call Summary & Next Steps</p>"
                    f"{f'<p style=margin:0 0 8px;font-size:14px;color:#333;>{summary}</p>' if summary else ''}"
                    f"{f'<ul style=margin:0;padding-left:20px;color:#333;font-size:14px;>{ai_li}</ul>' if ai_li else ''}</div>")
            body = (
                "<!doctype html><html><body style='margin:0;background:#f6f7f9;'>"
                "<div style='max-width:560px;margin:24px auto;padding:24px;background:#fff;"
                "border:1px solid #e6e8eb;border-radius:12px;font-family:-apple-system,Segoe UI,Roboto,Arial,sans-serif;color:#1a1a1a;'>"
                f"<p style='font-size:16px;margin:0 0 16px;'>Hi {first},</p>"
                "<p style='font-size:15px;line-height:1.6;margin:0 0 16px;'>Great speaking with you! "
                "As promised, here's what you needed:</p>"
                + "".join(blocks) +
                "<p style='font-size:15px;line-height:1.6;margin:16px 0 0;'>Just reply if you have any questions.</p>"
                f"<p style='font-size:15px;margin:12px 0 0;color:#555;'>— {(agent_data or {}).get('name', 'Team')}</p>"
                "</div></body></html>")
            subject = ("Your meeting details" if meeting and not materials else
                       "The information you asked for" if materials and not meeting else
                       f"Follow-up from {(agent_data or {}).get('name', 'ScatterStudio')} call")
            res = _int.execute_tool(owner_id, "send_email",
                                    {"to": lead_email, "subject": subject, "body": body})
            ok = bool((res or {}).get("ok"))
            bits = []
            if meeting:
                bits.append("meeting invite")
            if materials:
                bits.append("requested materials")
            if action_items:
                bits.append("action items")
            actions.append({"type": "email", "status": "ok" if ok else "failed",
                            "label": f"Emailed {lead_email}",
                            "detail": (res or {}).get("error", "") if not ok else (" + ".join(bits) or "summary")})
            logger.info(f"[PostCall] email to {lead_email}: ok={ok} ({' + '.join(bits)})")
        except Exception as e:
            logger.warning(f"[PostCall] email failed: {e}")
    elif not lead_email:
        logger.info("[PostCall] email skipped — no lead email captured")
    elif not has_actionable:
        logger.info(f"[PostCall] email skipped — no actionable items to deliver to {lead_email}")
    return actions


@api_router.post("/calls/{call_id}/analyze")
async def analyze_call_endpoint(call_id: str, background_tasks: BackgroundTasks):
    doc_ref = db.collection("calls").document(call_id)
    doc = doc_ref.get()
    if not doc.exists:
        raise HTTPException(status_code=404, detail="Call not found")
    
    call_data = doc.to_dict() or {}
    owner_id = call_data.get("owner_id")
    if not owner_id:
        raise HTTPException(status_code=400, detail="Owner ID missing in call log")
        
    background_tasks.add_task(_process_async_analysis, call_id, call_data)
    return {"status": "queued"}


# ----- Live Call Command Center -----
# The agent worker heartbeats each in-flight call into calls/{id} (live=True,
# rolling transcript, live_updated_at). These endpoints let the dashboard
# war-room (1) list live calls, (2) join the LiveKit room as a HIDDEN
# listen-only supervisor, and (3) "whisper" instructions the agent obeys
# mid-call (coach = invisible steering, say = speak verbatim).

class WhisperIn(BaseModel):
    text: str
    mode: str = "coach"  # "coach" steers the LLM invisibly; "say" speaks verbatim


@api_router.get("/calls/live")
def list_live_calls(user: dict = Depends(get_current_user)):
    docs = (db.collection("calls")
            .where(filter=FieldFilter("owner_id", "==", user["id"]))
            .where(filter=FieldFilter("live", "==", True))
            .limit(50).stream())
    out = []
    stale_cutoff = datetime.now(timezone.utc) - timedelta(seconds=45)
    for d in docs:
        c = d.to_dict() or {}
        # Drop stale "live" docs from crashed workers (heartbeat is ~2.5s).
        try:
            hb = datetime.fromisoformat((c.get("live_updated_at") or "").replace("Z", "+00:00"))
            if hb < stale_cutoff:
                continue
        except Exception:
            continue
        c.pop("transcript", None)  # detail poll fetches it
        out.append(c)
    out.sort(key=lambda x: x.get("started_at", ""), reverse=True)
    return out


@api_router.post("/calls/{call_id}/whisper")
def whisper_to_call(call_id: str, payload: WhisperIn, user: dict = Depends(get_current_user)):
    doc = db.collection("calls").document(call_id).get()
    if not doc.exists or doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=404, detail="Call not found")
    if not doc.to_dict().get("live"):
        raise HTTPException(status_code=400, detail="Call is no longer live")
    text = (payload.text or "").strip()
    if not text:
        raise HTTPException(status_code=400, detail="Whisper text is required")
    wid = uuid.uuid4().hex
    db.collection("calls").document(call_id).collection("whispers").document(wid).set({
        "id": wid, "text": text[:500],
        "mode": payload.mode if payload.mode in ("coach", "say") else "coach",
        "delivered": False, "created_at": now_iso(),
        "by": user.get("name") or user["id"],
    })
    return {"ok": True, "id": wid}


@api_router.get("/calls/{call_id}/listen-token")
def call_listen_token(call_id: str, user: dict = Depends(get_current_user)):
    """Mint a HIDDEN, subscribe-only LiveKit token so the owner can silently
    listen to a live call from the dashboard. The agent worker ignores
    supervisor_* identities, so joining/leaving never disturbs the call."""
    doc = db.collection("calls").document(call_id).get()
    if not doc.exists or doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=404, detail="Call not found")
    c = doc.to_dict()
    room_name = c.get("room_name")
    if not room_name:
        raise HTTPException(status_code=400, detail="This call has no live room")
    api_key = os.environ.get("LIVEKIT_API_KEY")
    api_secret = os.environ.get("LIVEKIT_API_SECRET")
    livekit_url = os.environ.get("LIVEKIT_URL")
    if not (api_key and api_secret and livekit_url):
        raise HTTPException(status_code=500, detail="LiveKit not configured")
    token = (
        livekit_api.AccessToken(api_key, api_secret)
        .with_identity(f"supervisor_{user['id'][:12]}_{uuid.uuid4().hex[:6]}")
        .with_name("Supervisor")
        .with_grants(livekit_api.VideoGrants(
            room_join=True, room=room_name,
            can_publish=False, can_subscribe=True, hidden=True,
        ))
    )
    return {"token": token.to_jwt(), "url": livekit_url, "room": room_name}


# ----- Call Logs -----
@api_router.get("/calls", response_model=List[CallLog])
def list_calls(
    user: dict = Depends(get_current_user),
    agent_id: Optional[str] = None,
    sentiment: Optional[str] = None,
    lead_tag: Optional[str] = None,
    search: Optional[str] = None,
):
    cache_key = f"calls:{user['id']}:{agent_id or ''}:{sentiment or ''}:{lead_tag or ''}:{search or ''}"
    cached = _cache_get(cache_key)
    if cached is not None:
        return cached
    query = db.collection("calls").where(filter=FieldFilter("owner_id", "==", user["id"]))
    if agent_id:
        query = query.where(filter=FieldFilter("agent_id", "==", agent_id))
    if sentiment:
        query = query.where(filter=FieldFilter("sentiment", "==", sentiment))
    if lead_tag:
        query = query.where(filter=FieldFilter("lead_tag", "==", lead_tag))
    # Cap the list and drop the heavy `transcript` field — the list view never
    # renders it (only /calls/{id} does). This keeps the payload small & fast.
    query = query.limit(500)
    docs = query.stream()
    results = []
    for doc in docs:
        d = doc.to_dict()
        d.pop("transcript", None)
        results.append(d)
    if search:
        s = search.lower()
        results = [
            r for r in results
            if s in (r.get("phone_number") or "").lower()
            or s in (r.get("contact_name") or "").lower()
            or s in (r.get("summary") or "").lower()
        ]
    results.sort(key=lambda x: x.get("started_at", ""), reverse=True)
    # 15s cache — call rows change slowly; the detail view fetches fresh anyway.
    return _cache_set(cache_key, results, ttl=15)


@api_router.get("/calls/{call_id}", response_model=CallLog)
def get_call(call_id: str, user: dict = Depends(get_current_user)):
    doc = db.collection("calls").document(call_id).get()
    if not doc.exists or doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=404, detail="Call not found")
    return doc.to_dict()


# ----- Leads (captured automatically from calls) -----
@api_router.get("/leads", response_model=List[Lead])
def list_leads(
    user: dict = Depends(get_current_user),
    agent_id: Optional[str] = None,
    status: Optional[str] = None,
    search: Optional[str] = None,
):
    q = db.collection("leads").where(filter=FieldFilter("owner_id", "==", user["id"]))
    if agent_id:
        q = q.where(filter=FieldFilter("agent_id", "==", agent_id))
    if status:
        q = q.where(filter=FieldFilter("status", "==", status))
    results = [d.to_dict() for d in q.stream()]
    if search:
        s = search.lower()
        results = [
            r for r in results
            if s in (r.get("name") or "").lower()
            or s in (r.get("phone") or "").lower()
            or s in (r.get("email") or "").lower()
            or s in (r.get("company") or "").lower()
        ]
    # Sort by score desc then most-recent (best-effort; created_at is a Timestamp).
    results.sort(key=lambda x: x.get("score", 0), reverse=True)
    return results


@api_router.delete("/leads/{lead_id}")
def delete_lead(lead_id: str, user: dict = Depends(get_current_user)):
    doc = db.collection("leads").document(lead_id).get()
    if doc.exists and doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=403, detail="Not allowed")
    db.collection("leads").document(lead_id).delete()
    return {"ok": True}


# ----- Notifications (bell) -----
@api_router.get("/notifications")
def get_notifications(user: dict = Depends(get_current_user)):
    """Urgent items for the bell: HOT leads to call now, ANGRY/negative-sentiment
    calls to review, and post-call actions (emails sent, meetings booked +
    upcoming-meeting reminders). Cached briefly per user."""
    uid = user["id"]
    cached = _cache_get(f"notifs:{uid}")
    if cached is not None:
        return cached

    notes = []
    now = datetime.now(timezone.utc)

    # 1. Hot leads — "call them now"
    try:
        for d in (db.collection("leads").where(filter=FieldFilter("owner_id", "==", uid))
                  .where(filter=FieldFilter("status", "==", "hot")).limit(15).stream()):
            l = d.to_dict()
            notes.append({
                "id": f"lead_{d.id}", "type": "hot_lead", "severity": "high",
                "icon": "🔥", "title": f"Hot lead: {l.get('name') or l.get('phone') or 'Unknown'}",
                "body": (l.get("intent") or l.get("message") or "")[:120],
                "phone": l.get("phone", ""), "link": "/app/leads",
                "ts": l.get("updated_at"),
            })
    except Exception as e:
        logger.warning(f"[Notif] leads query: {e}")

    # 2. Negative-sentiment calls — "needs attention"
    try:
        for d in (db.collection("calls").where(filter=FieldFilter("owner_id", "==", uid))
                  .where(filter=FieldFilter("sentiment", "==", "negative")).limit(15).stream()):
            c = d.to_dict()
            notes.append({
                "id": f"call_{d.id}", "type": "angry_call", "severity": "high",
                "icon": "⚠️", "title": f"Unhappy caller: {c.get('contact_name') or c.get('phone_number') or 'Unknown'}",
                "body": (c.get("summary") or "Negative sentiment detected on this call.")[:120],
                "phone": c.get("phone_number", ""), "link": "/app/call-logs",
                "ts": c.get("started_at"),
            })
    except Exception as e:
        logger.warning(f"[Notif] calls query: {e}")

    # 3. Post-call actions + upcoming-meeting reminders (from calls[].actions).
    try:
        recent = sorted(
            [c.to_dict() for c in db.collection("calls").where(filter=FieldFilter("owner_id", "==", uid)).limit(80).stream()],
            key=lambda x: x.get("started_at", ""), reverse=True)[:30]
        for c in recent:
            for a in (c.get("actions") or []):
                if a.get("status") != "ok":
                    continue
                if a.get("type") == "email":
                    notes.append({"id": f"email_{c.get('id')}", "type": "email_sent", "severity": "info",
                                  "icon": "📧", "title": a.get("label", "Email sent"),
                                  "body": a.get("detail", ""), "link": "/app/call-logs", "ts": c.get("started_at")})
                elif a.get("type") == "calendar":
                    notes.append({"id": f"cal_{c.get('id')}", "type": "meeting", "severity": "info",
                                  "icon": "📅", "title": a.get("label", "Meeting booked"),
                                  "body": a.get("detail", ""), "link": "/app/call-logs", "ts": c.get("started_at")})
    except Exception as e:
        logger.warning(f"[Notif] actions query: {e}")

    # Cap + shape
    high = [n for n in notes if n.get("severity") == "high"]
    info = [n for n in notes if n.get("severity") != "high"]
    out = {"count": len(high), "total": len(notes), "items": (high + info)[:25],
           "generated_at": now.isoformat()}
    return _cache_set(f"notifs:{uid}", out, ttl=20)


# ----- Telephony self-diagnostic -----
@api_router.get("/telephony/diagnostic")
def telephony_diagnostic(user: dict = Depends(get_current_user)):
    """Report which env vars / SIP resources are configured so the user can
    self-diagnose 'calls fail before audio' without server access. No secrets."""
    env = {k: bool(os.environ.get(k)) for k in [
        "LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET",
        "LIVEKIT_SIP_TRUNK_ID", "VOBIZ_SIP_DOMAIN", "VOBIZ_SIP_USERNAME",
        "VOBIZ_AUTH_ID", "VOBIZ_FROM_NUMBER",
        "GROQ_API_KEY", "SARVAM_API_KEY", "NVIDIA_API_KEY", "GEMINI_API_KEY",
    ]}
    trunks = list(db.collection("sip_trunks").where(filter=FieldFilter("owner_id", "==", user["id"])).stream())
    numbers = list(db.collection("phone_numbers").where(filter=FieldFilter("owner_id", "==", user["id"])).stream())
    issues = []
    if not all([env["LIVEKIT_URL"], env["LIVEKIT_API_KEY"], env["LIVEKIT_API_SECRET"]]):
        issues.append("LiveKit env vars missing — web + phone calls will fail.")
    if not env["LIVEKIT_SIP_TRUNK_ID"] and not trunks and not (env["VOBIZ_SIP_DOMAIN"]):
        issues.append("No SIP trunk configured — outbound phone calls cannot dial. "
                      "Set VOBIZ_SIP_DOMAIN + creds or create a trunk in Phone Numbers.")
    if not numbers and not env["VOBIZ_FROM_NUMBER"]:
        issues.append("No caller-ID number — assign a phone number to an agent or set VOBIZ_FROM_NUMBER.")
    if not env["GROQ_API_KEY"]:
        issues.append("GROQ_API_KEY missing — the agent's LLM and post-call analysis will not work.")
    if not env["SARVAM_API_KEY"]:
        issues.append("SARVAM_API_KEY missing — Sarvam STT/TTS will not work.")
    return {
        "env": env,
        "sip_trunks": len(trunks),
        "phone_numbers": len(numbers),
        "issues": issues,
        "status": "ok" if not issues else "degraded",
    }


@api_router.get("/gemini/live-models")
async def gemini_live_models():
    """PUBLIC — list the Gemini models that support realtime (bidiGenerateContent)
    for the configured GEMINI_API_KEY, so we can pick a model name that actually
    exists. Open this URL in a browser to see the valid Gemini Live model ids."""
    key = os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY", "")
    if not key:
        return {"error": "GEMINI_API_KEY not set on the server"}
    try:
        async with httpx.AsyncClient(timeout=20.0) as client:
            r = await client.get(
                f"https://generativelanguage.googleapis.com/v1beta/models?key={key}&pageSize=1000")
            r.raise_for_status()
            models = r.json().get("models", [])
    except Exception as e:
        return {"error": f"query failed: {e}"}
    live, flash = [], []
    for m in models:
        name = m.get("name", "").replace("models/", "")
        methods = m.get("supportedGenerationMethods", [])
        if "bidiGenerateContent" in methods:
            live.append(name)
        if "live" in name or "flash" in name:
            flash.append(name)
    return {"realtime_bidi_models": sorted(live),
            "current_env_default": os.environ.get("GEMINI_LIVE_MODEL", "(unset)"),
            "hint": "Set GEMINI_LIVE_MODEL to one of realtime_bidi_models on the Space, then Restart."}


@api_router.get("/integrations/diagnostic")
def integrations_diagnostic():
    """PUBLIC (no auth) OAuth self-check — open this URL in a browser to debug
    'Error 401: invalid_client'. Reports which OAuth client IDs the server sees
    (masked, no secrets), the OAuth redirect base it will use, and the exact
    redirect URIs to register in each provider console. If a client_id shows
    present=false, that provider's Secret isn't set on the Space → that's the 401.
    """
    import integrations as _int

    def _probe(env_key):
        cid = (os.environ.get(env_key) or "").strip()
        return {"present": bool(cid),
                "value_preview": (cid[:14] + "…" + cid[-6:]) if len(cid) > 24 else cid,
                "looks_trimmed": cid == (os.environ.get(env_key) or "")}

    base = _int._redirect_base()
    google_cid = _probe("GOOGLE_CLIENT_ID")
    providers = {}
    for pid in ("gmail", "google_calendar", "google_sheets", "hubspot", "zoho", "outlook"):
        # Resolve the effective client_id the same way the OAuth flow does.
        try:
            cid, sec = _int._client_creds(pid, {})
        except Exception:
            cid, sec = None, None
        providers[pid] = {
            "client_id_present": bool(cid),
            "client_secret_present": bool(sec),
            "redirect_uri": f"{base}/api/integrations/{pid}/callback" if base else "(OAUTH_REDIRECT_BASE not set)",
        }

    return {
        "oauth_redirect_base": base or "(EMPTY — set OAUTH_REDIRECT_BASE)",
        "GOOGLE_CLIENT_ID": google_cid,
        "GOOGLE_CLIENT_SECRET_present": bool((os.environ.get("GOOGLE_CLIENT_SECRET") or "").strip()),
        "providers": providers,
        "hint": ("If a Google provider shows client_id_present=false, add "
                 "GOOGLE_CLIENT_ID + GOOGLE_CLIENT_SECRET to the Space Secrets and Restart. "
                 "Then register the shown redirect_uri values in the Google console."),
    }


# ----- Phone Numbers -----
@api_router.get("/phone-numbers", response_model=List[PhoneNumber])
def list_numbers(user: dict = Depends(get_current_user)):
    cached = _cache_get(f"phone_numbers:{user['id']}")
    if cached is not None:
        return cached
    docs = db.collection("phone_numbers").where(filter=FieldFilter("owner_id", "==", user["id"])).stream()
    return _cache_set(f"phone_numbers:{user['id']}", [doc.to_dict() for doc in docs], ttl=30)


@api_router.post("/phone-numbers", response_model=PhoneNumber)
def add_number(payload: dict, user: dict = Depends(get_current_user)):
    number = (payload.get("number") or "").strip()
    if not number:
        raise HTTPException(status_code=400, detail="Phone number is required")
    pid = str(uuid.uuid4())
    doc_data = {
        "id": pid,
        "number": number,
        "provider": "Vobiz",
        "label": payload.get("label", "New Number"),
        "assigned_agent_id": payload.get("assigned_agent_id"),
        "status": "active",
        "sip_domain": payload.get("sip_domain"),
        "sip_username": payload.get("sip_username"),
        "sip_password": payload.get("sip_password"),
        "owner_id": user["id"],
    }
    db.collection("phone_numbers").document(pid).set(doc_data)
    _cache_invalidate_user(user["id"], "phone_numbers")
    return {k: v for k, v in doc_data.items() if k != "owner_id"}


@api_router.delete("/phone-numbers/{pid}")
def delete_number(pid: str, user: dict = Depends(get_current_user)):
    doc = db.collection("phone_numbers").document(pid).get()
    if doc.exists and doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=403, detail="Not allowed")
    db.collection("phone_numbers").document(pid).delete()
    _cache_invalidate_user(user["id"], "phone_numbers")
    return {"ok": True}


# ----- Knowledge Base (real file upload + chunking) -----
@api_router.get("/knowledge", response_model=List[KnowledgeDoc])
def list_knowledge(user: dict = Depends(get_current_user)):
    cached = _cache_get(f"knowledge:{user['id']}")
    if cached is not None:
        return cached
    docs = db.collection("knowledge").where(filter=FieldFilter("owner_id", "==", user["id"])).stream()
    results = []
    for doc in docs:
        try:
            d = doc.to_dict() or {}
            # Ensure required fields exist and types are safe before Pydantic validation
            d.setdefault("name", "")
            d.setdefault("size_kb", 0)
            d.setdefault("pages", 0)
            d.setdefault("status", "ready")
            d.setdefault("uploaded_at", "")
            d.setdefault("chunks", 0)
            d["size_kb"] = float(d["size_kb"] or 0)
            d["pages"] = int(d["pages"] or 0)
            d["chunks"] = int(d["chunks"] or 0)
            results.append(d)
        except Exception as e:
            logger.warning(f"Skipping malformed KB doc {doc.id}: {e}")
    return _cache_set(f"knowledge:{user['id']}", results, ttl=30)


@api_router.post("/knowledge/upload", response_model=KnowledgeDoc)
async def upload_knowledge(file: UploadFile = File(...), user: dict = Depends(get_current_user)):
    content = await file.read()
    size_kb = max(1, len(content) // 1024)
    name = file.filename or "Document"
    ext = (name.rsplit(".", 1)[-1] or "").lower()

    text = ""
    pages = 1
    try:
        if ext == "pdf":
            text, pages = _extract_pdf_text(content)
        elif ext in ("txt", "md", "csv"):
            text = content.decode("utf-8", errors="ignore")
            pages = max(1, len(text) // 3000)
        elif ext == "docx":
            text, pages = _extract_docx_text(content)
        else:
            try:
                text = content.decode("utf-8", errors="ignore")
            except Exception:
                text = ""
    except Exception as e:
        logger.error(f"KB parse failed: {e}")

    kid = str(uuid.uuid4())
    chunks = _chunk_text(text, target=900)
    doc_data = {
        "id": kid,
        "name": name,
        "size_kb": size_kb,
        "pages": pages,
        "status": "ready" if chunks else "failed",
        "uploaded_at": now_iso(),
        "owner_id": user["id"],
        "chunks": len(chunks),
    }
    db.collection("knowledge").document(kid).set(doc_data)
    # Store chunks in subcollection
    for i, ch in enumerate(chunks):
        vector = _get_embedding(ch, input_type="passage")
        db.collection("knowledge").document(kid).collection("chunks").document(str(i)).set({
            "i": i, 
            "text": ch,
            "vector": vector
        })

    _cache_invalidate_user(user["id"], "knowledge")
    return {k: v for k, v in doc_data.items() if k != "owner_id"}

class URLUploadIn(BaseModel):
    url: str

import urllib.parse
import requests
from bs4 import BeautifulSoup

def _crawl_and_embed(start_url: str, doc_id: str, owner_id: str, max_pages: int = 20):
    visited = set()
    queue = [start_url]
    base_domain = urllib.parse.urlparse(start_url).netloc
    
    total_chunks = 0
    pages_crawled = 0
    total_bytes = 0
    
    while queue and pages_crawled < max_pages:
        current_url = queue.pop(0)
        if current_url in visited:
            continue
            
        visited.add(current_url)
        pages_crawled += 1
        
        try:
            r = requests.get(current_url, timeout=10)
            if r.status_code != 200:
                continue
                
            soup = BeautifulSoup(r.text, 'html.parser')
            
            for script in soup(["script", "style"]):
                script.decompose()
            text = soup.get_text(separator=' ', strip=True)
            
            if text:
                total_bytes += len(text.encode("utf-8"))
                chunks = _chunk_text(text, target=900)
                for ch in chunks:
                    vector = _get_embedding(ch, input_type="passage")
                    db.collection("knowledge").document(doc_id).collection("chunks").document(str(total_chunks)).set({
                        "i": total_chunks,
                        "text": ch,
                        "vector": vector,
                        "source": current_url
                    })
                    total_chunks += 1
                    
            for link in soup.find_all('a', href=True):
                href = link['href']
                absolute_url = urllib.parse.urljoin(current_url, href)
                parsed = urllib.parse.urlparse(absolute_url)
                
                clean_url = urllib.parse.urlunparse((parsed.scheme, parsed.netloc, parsed.path, parsed.params, parsed.query, ''))
                
                if parsed.netloc == base_domain and clean_url not in visited and clean_url not in queue:
                    if parsed.scheme in ("http", "https"):
                        queue.append(clean_url)
                        
        except Exception as e:
            logger.error(f"Crawler failed on {current_url}: {e}")
            
    db.collection("knowledge").document(doc_id).update({
        "status": "ready" if total_chunks > 0 else "failed",
        "chunks": total_chunks,
        "pages": pages_crawled,
        "size_kb": round(total_bytes / 1024, 1)
    })
    _cache_invalidate_user(owner_id, "knowledge")

@api_router.post("/knowledge/upload-url", response_model=KnowledgeDoc)
def upload_url_knowledge(payload: URLUploadIn, background_tasks: BackgroundTasks, user: dict = Depends(get_current_user)):
    url = payload.url.strip()
    if not url.startswith("http"):
        url = "https://" + url
        
    kid = str(uuid.uuid4())
    doc_data = {
        "id": kid,
        "name": url,
        "size_kb": 0,
        "pages": 0,
        "status": "indexing",
        "uploaded_at": now_iso(),
        "owner_id": user["id"],
        "chunks": 0,
    }
    db.collection("knowledge").document(kid).set(doc_data)
    _cache_invalidate_user(user["id"], "knowledge")
    
    background_tasks.add_task(_crawl_and_embed, url, kid, user["id"], 20)
    return {k: v for k, v in doc_data.items() if k != "owner_id"}


@api_router.post("/knowledge", response_model=KnowledgeDoc)
def add_knowledge_meta(payload: dict, user: dict = Depends(get_current_user)):
    """Metadata-only entry (legacy). Prefer /knowledge/upload for actual content."""
    kid = str(uuid.uuid4())
    doc_data = {
        "id": kid,
        "name": payload.get("name", "Document"),
        "size_kb": int(payload.get("size_kb", 0)),
        "pages": int(payload.get("pages", 1)),
        "status": "processing",
        "uploaded_at": now_iso(),
        "owner_id": user["id"],
        "chunks": 0,
    }
    db.collection("knowledge").document(kid).set(doc_data)
    _cache_invalidate_user(user["id"], "knowledge")
    return {k: v for k, v in doc_data.items() if k != "owner_id"}


@api_router.get("/knowledge/{kid}/chunks")
def get_knowledge_chunks(kid: str, user: dict = Depends(get_current_user)):
    """Return all text chunks for a knowledge document (without vectors) for preview."""
    doc = db.collection("knowledge").document(kid).get()
    if not doc.exists:
        raise HTTPException(status_code=404, detail="Document not found")
    if doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=403, detail="Not allowed")
    chunks = []
    for c in db.collection("knowledge").document(kid).collection("chunks").order_by("i").stream():
        d = c.to_dict()
        chunks.append({"i": d.get("i", 0), "text": d.get("text", ""), "source": d.get("source", "")})
    return chunks


@api_router.delete("/knowledge/{kid}")
def delete_knowledge(kid: str, user: dict = Depends(get_current_user)):
    doc = db.collection("knowledge").document(kid).get()
    if not doc.exists:
        return {"ok": True}
    if doc.to_dict().get("owner_id") != user["id"]:
        raise HTTPException(status_code=403, detail="Not allowed")
    # Delete chunk subcollection
    for sub in db.collection("knowledge").document(kid).collection("chunks").stream():
        sub.reference.delete()
    db.collection("knowledge").document(kid).delete()
    _cache_invalidate_user(user["id"], "knowledge")
    return {"ok": True}


def _extract_pdf_text(data: bytes) -> tuple[str, int]:
    try:
        from pypdf import PdfReader  # type: ignore
    except ImportError:
        try:
            from PyPDF2 import PdfReader  # type: ignore
        except ImportError:
            logger.warning("pypdf not installed; PDF parsing skipped")
            return "", 1
    reader = PdfReader(io.BytesIO(data))
    pages = len(reader.pages)
    out = []
    for p in reader.pages:
        try:
            out.append(p.extract_text() or "")
        except Exception:
            pass
    return "\n".join(out), pages


def _extract_docx_text(data: bytes) -> tuple[str, int]:
    try:
        from docx import Document  # python-docx
    except ImportError:
        logger.warning("python-docx not installed")
        return "", 1
    d = Document(io.BytesIO(data))
    text = "\n".join(p.text for p in d.paragraphs if p.text)
    return text, max(1, len(text) // 3000)


def _chunk_text(text: str, target: int = 900) -> List[str]:
    text = (text or "").strip()
    if not text:
        return []
    paras = re.split(r"\n\s*\n", text)
    chunks: List[str] = []
    buf = ""
    for p in paras:
        if len(buf) + len(p) + 2 > target and buf:
            chunks.append(buf.strip())
            buf = p
        else:
            buf += ("\n\n" + p) if buf else p
    if buf.strip():
        chunks.append(buf.strip())
    return chunks


# ----- Voices & TTS -----
VOICE_CATALOG = {
    "Sarvam": [
        {"id": "shreya", "label": "Shreya (Female)", "gender": "female"},
        {"id": "ishita", "label": "Ishita (Female)", "gender": "female"},
        {"id": "priya", "label": "Priya (Female)", "gender": "female"},
        {"id": "suhani", "label": "Suhani (Female)", "gender": "female"},
        {"id": "ritu", "label": "Ritu (Female)", "gender": "female"},
        {"id": "pooja", "label": "Pooja (Female)", "gender": "female"},
        {"id": "simran", "label": "Simran (Female)", "gender": "female"},
        {"id": "kavya", "label": "Kavya (Female)", "gender": "female"},
        {"id": "neha", "label": "Neha (Female)", "gender": "female"},
        {"id": "roopa", "label": "Roopa (Female)", "gender": "female"},
        {"id": "tanya", "label": "Tanya (Female)", "gender": "female"},
        {"id": "shruti", "label": "Shruti (Female)", "gender": "female"},
        {"id": "kavitha", "label": "Kavitha (Female)", "gender": "female"},
        {"id": "rupali", "label": "Rupali (Female)", "gender": "female"},
        {"id": "shubh", "label": "Shubh (Male)", "gender": "male"},
        {"id": "manan", "label": "Manan (Male)", "gender": "male"},
        {"id": "amit", "label": "Amit (Male)", "gender": "male"},
        {"id": "rahul", "label": "Rahul (Male)", "gender": "male"},
        {"id": "rohan", "label": "Rohan (Male)", "gender": "male"},
        {"id": "aditya", "label": "Aditya (Male)", "gender": "male"},
        {"id": "dev", "label": "Dev (Male)", "gender": "male"},
        {"id": "ratan", "label": "Ratan (Male)", "gender": "male"},
        {"id": "varun", "label": "Varun (Male)", "gender": "male"},
        {"id": "sumit", "label": "Sumit (Male)", "gender": "male"},
        {"id": "kabir", "label": "Kabir (Male)", "gender": "male"},
        {"id": "aayan", "label": "Aayan (Male)", "gender": "male"},
        {"id": "ashutosh", "label": "Ashutosh (Male)", "gender": "male"},
        {"id": "advait", "label": "Advait (Male)", "gender": "male"},
        {"id": "anand", "label": "Anand (Male)", "gender": "male"},
        {"id": "tarun", "label": "Tarun (Male)", "gender": "male"},
        {"id": "sunny", "label": "Sunny (Male)", "gender": "male"},
        {"id": "mani", "label": "Mani (Male)", "gender": "male"},
        {"id": "gokul", "label": "Gokul (Male)", "gender": "male"},
        {"id": "vijay", "label": "Vijay (Male)", "gender": "male"},
        {"id": "mohit", "label": "Mohit (Male)", "gender": "male"},
        {"id": "rehan", "label": "Rehan (Male)", "gender": "male"},
        {"id": "soham", "label": "Soham (Male)", "gender": "male"},
    ],
    "ElevenLabs": [
        {"id": "21m00Tcm4TlvDq8ikWAM", "label": "Rachel (Female)", "gender": "female"},
        {"id": "pNInz6obpgDQGcFmaJgB", "label": "Adam (Male)", "gender": "male"},
        {"id": "EXAVITQu4vr4xnSDxMaL", "label": "Bella (Female)", "gender": "female"},
        {"id": "TxGEqnHWrfWFTfGW9XjX", "label": "Josh (Male)", "gender": "male"},
    ],
    "OpenAI": [
        {"id": "alloy", "label": "Alloy (Neutral)", "gender": "neutral"},
        {"id": "echo", "label": "Echo (Male)", "gender": "male"},
        {"id": "fable", "label": "Fable (Male)", "gender": "male"},
        {"id": "onyx", "label": "Onyx (Male)", "gender": "male"},
        {"id": "nova", "label": "Nova (Female)", "gender": "female"},
        {"id": "shimmer", "label": "Shimmer (Female)", "gender": "female"},
    ],
}


@api_router.get("/voices")
def list_voices():
    return VOICE_CATALOG


class TTSPreviewIn(BaseModel):
    voice_id: str = "shreya"
    text: str = "Namaste! Main aapki kaise madad kar sakti hoon?"
    speed: float = 1.0
    provider: str = "Sarvam"


@api_router.post("/tts/preview")
async def tts_preview(payload: TTSPreviewIn, user: dict = Depends(get_current_user)):
    return await _route_tts(payload)


@api_router.post("/tts/test")
async def tts_test(payload: TTSPreviewIn, user: dict = Depends(get_current_user)):
    return await _route_tts(payload)


async def _route_tts(payload: TTSPreviewIn):
    provider = (payload.provider or "Sarvam").lower()
    if provider in ("elevenlabs", "eleven"):
        return await _call_elevenlabs_tts(payload)
    if provider in ("openai", "whisper"):
        return await _call_openai_tts(payload)
    return await _call_sarvam_tts(payload)


async def _call_sarvam_tts(payload: TTSPreviewIn):
    key = os.environ.get("SARVAM_API_KEY")
    if not key:
        raise HTTPException(status_code=500, detail="SARVAM_API_KEY not configured")
    body = {
        "inputs": [payload.text],
        "target_language_code": "hi-IN",
        "speaker": payload.voice_id,
        "model": "bulbul:v3",
        "speed": float(payload.speed),
        "enable_preprocessing": True,
    }
    try:
        async with httpx.AsyncClient(timeout=20.0) as client:
            r = await client.post(
                "https://api.sarvam.ai/text-to-speech",
                headers={"api-subscription-key": key, "Content-Type": "application/json"},
                json=body,
            )
            if r.status_code != 200:
                logger.error(f"Sarvam TTS error {r.status_code}: {r.text[:200]}")
                raise HTTPException(status_code=502, detail=f"Sarvam {r.status_code}: {r.text[:200]}")
            audios = r.json().get("audios") or []
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"TTS error: {type(e).__name__}: {e}")
    if not audios:
        raise HTTPException(status_code=502, detail="No audio returned")
    return StreamingResponse(io.BytesIO(base64.b64decode(audios[0])), media_type="audio/wav")


async def _call_elevenlabs_tts(payload: TTSPreviewIn):
    key = os.environ.get("ELEVENLABS_API_KEY")
    if not key:
        raise HTTPException(status_code=500, detail="ELEVENLABS_API_KEY not configured")
    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            r = await client.post(
                f"https://api.elevenlabs.io/v1/text-to-speech/{payload.voice_id}",
                headers={"xi-api-key": key, "Content-Type": "application/json", "Accept": "audio/mpeg"},
                json={
                    "text": payload.text,
                    "model_id": "eleven_multilingual_v2",
                    "voice_settings": {"stability": 0.5, "similarity_boost": 0.75},
                },
            )
            if r.status_code != 200:
                raise HTTPException(status_code=502, detail=f"ElevenLabs {r.status_code}: {r.text[:200]}")
            return StreamingResponse(io.BytesIO(r.content), media_type="audio/mpeg")
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"ElevenLabs error: {e}")


async def _call_openai_tts(payload: TTSPreviewIn):
    key = os.environ.get("OPENAI_API_KEY")
    if not key:
        raise HTTPException(status_code=500, detail="OPENAI_API_KEY not configured")
    try:
        o_vid = payload.voice_id.lower() if payload.voice_id.lower() in ("alloy", "echo", "fable", "onyx", "nova", "shimmer") else "alloy"
        async with httpx.AsyncClient(timeout=30.0) as client:
            r = await client.post(
                "https://api.openai.com/v1/audio/speech",
                headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
                json={
                    "model": "tts-1",
                    "input": payload.text,
                    "voice": o_vid,
                    "speed": float(payload.speed),
                },
            )
            if r.status_code != 200:
                raise HTTPException(status_code=502, detail=f"OpenAI TTS error {r.status_code}: {r.text[:200]}")
            return StreamingResponse(io.BytesIO(r.content), media_type="audio/mpeg")
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"OpenAI TTS error: {e}")


# ----- Settings -----
class ProfileUpdateIn(BaseModel):
    name: Optional[str] = None
    workspace: Optional[str] = None


@api_router.put("/settings/profile")
def update_profile(payload: ProfileUpdateIn, user: dict = Depends(get_current_user)):
    update_data: Dict[str, Any] = {}
    if payload.name is not None:
        n = payload.name.strip()
        if not (1 <= len(n) <= 100):
            raise HTTPException(status_code=400, detail="Name must be 1-100 characters")
        update_data["name"] = n
    if payload.workspace is not None:
        w = payload.workspace.strip()
        if not (1 <= len(w) <= 60):
            raise HTTPException(status_code=400, detail="Workspace must be 1-60 characters")
        update_data["workspace"] = w
    if update_data:
        db.collection("users").document(user["id"]).update(update_data)
        _cache_invalidate_user(user["id"])
    return {"ok": True, **update_data}


class ProviderConnectIn(BaseModel):
    provider: str
    api_key: Optional[str] = None
    api_secret: Optional[str] = None
    enabled: bool = True


@api_router.get("/settings/providers")
def list_provider_integrations(user: dict = Depends(get_current_user)):
    cached = _cache_get(f"providers:{user['id']}")
    if cached is not None:
        return cached
    docs = db.collection("provider_integrations").where(filter=FieldFilter("owner_id", "==", user["id"])).stream()
    out = {d.id.split("__")[-1]: d.to_dict() for d in docs}
    return _cache_set(f"providers:{user['id']}", out, ttl=60)


@api_router.put("/settings/providers")
def upsert_provider_integration(payload: ProviderConnectIn, user: dict = Depends(get_current_user)):
    if not payload.provider:
        raise HTTPException(status_code=400, detail="provider is required")
    doc_id = f"{user['id']}__{payload.provider}"
    data = {
        "provider": payload.provider,
        "enabled": payload.enabled,
        "owner_id": user["id"],
        "updated_at": now_iso(),
        # Note: in production, encrypt these or use Secret Manager
        "has_key": bool(payload.api_key),
        "has_secret": bool(payload.api_secret),
    }
    if payload.api_key:
        data["api_key_masked"] = "•" * max(0, len(payload.api_key) - 4) + payload.api_key[-4:]
        data["api_key"] = payload.api_key
    if payload.api_secret:
        data["api_secret"] = payload.api_secret
    db.collection("provider_integrations").document(doc_id).set(data, merge=True)
    _cache_invalidate_user(user["id"], "providers")
    return {"ok": True, "provider": payload.provider, "enabled": payload.enabled}


@api_router.post("/settings/api-key/rotate")
def rotate_api_key(user: dict = Depends(get_current_user)):
    new_key = "sk-scatter-" + secrets.token_hex(20)
    db.collection("users").document(user["id"]).update({"api_key": new_key})
    _cache_invalidate_user(user["id"])
    return {"api_key": new_key}


# ----- Integrations (OAuth: Gmail / Outlook / HubSpot / Zoho + SMTP) -----
import integrations as _integrations
from fastapi.responses import HTMLResponse


@api_router.get("/integrations")
def list_integrations_route(user: dict = Depends(get_current_user)):
    """Catalog with per-provider connection status for the current user."""
    return {"integrations": _integrations.list_integrations(user["id"])}


@api_router.get("/integrations/{provider}/connect")
def integration_connect(provider: str, request: Request, user: dict = Depends(get_current_user)):
    """Begin OAuth — returns the provider consent URL for the frontend to open."""
    res = _integrations.build_authorize_url(user["id"], provider, str(request.base_url))
    if not res.get("ok"):
        raise HTTPException(status_code=400, detail=res.get("error", "Cannot start OAuth"))
    return {"redirect_url": res["redirect_url"]}


@api_router.get("/integrations/{provider}/callback")
def integration_callback(provider: str, code: str = "", state: str = ""):
    """OAuth redirect target (no auth — uid is carried in `state`). Exchanges the
    code for tokens, then returns a tiny page that notifies the opener tab."""
    res = _integrations.exchange_code(provider, code, state)
    ok = bool(res.get("ok"))
    msg = "Connected" if ok else (res.get("error") or "Connection failed")
    color = "#10b981" if ok else "#ef4444"
    html = (
        "<!doctype html><meta charset='utf-8'><title>" + msg + "</title>"
        "<style>body{font-family:Inter,system-ui,sans-serif;background:#0B0B10;color:#F5F5F7;"
        "display:grid;place-items:center;height:100vh;margin:0}</style>"
        "<div style='text-align:center'><h2 style='color:" + color + "'>"
        + ("&#10003; " if ok else "&#10007; ") + msg + "</h2>"
        "<p style='color:#A8A8B3'>You can close this tab.</p></div>"
        "<script>try{window.opener&&window.opener.postMessage({scatter_integration:'"
        + provider + "',connected:" + ("true" if ok else "false") + "},'*')}catch(e){};"
        "setTimeout(function(){window.close()},1200)</script>"
    )
    return HTMLResponse(content=html)


@api_router.post("/integrations/{provider}/disconnect")
def integration_disconnect(provider: str, user: dict = Depends(get_current_user)):
    _integrations.delete_cfg(user["id"], provider)
    return {"ok": True}


@api_router.post("/integrations/{provider}/save")
async def integration_save(provider: str, request: Request, user: dict = Depends(get_current_user)):
    """Save non-OAuth credentials (SMTP host/port/user/pass) or a power user's own
    OAuth client_id/secret."""
    data = await request.json()
    res = _integrations.save_manual(user["id"], provider, data or {})
    if not res.get("ok"):
        raise HTTPException(status_code=400, detail=res.get("error", "Save failed"))
    return res


# ----- Billing / Usage -----
@api_router.get("/billing/usage")
def billing_usage(user: dict = Depends(get_current_user)):
    cached = _cache_get(f"billing:{user['id']}")
    if cached is not None:
        return cached
    # Plan defaults — store under users/{uid}/plan in future
    plan = {"name": "Growth", "agents_limit": 5, "minutes_limit": 5000, "docs_limit": 50, "mrr": "₹24,999"}
    user_doc = db.collection("users").document(user["id"]).get()
    if user_doc.exists:
        ud = user_doc.to_dict()
        if ud.get("plan"):
            plan.update(ud["plan"])

    agents_count = len(list(db.collection("agents").where(filter=FieldFilter("owner_id", "==", user["id"])).stream()))
    docs_count = len(list(db.collection("knowledge").where(filter=FieldFilter("owner_id", "==", user["id"])).stream()))

    # Sum minutes from this month's calls
    month_start = datetime.now(timezone.utc).replace(day=1, hour=0, minute=0, second=0, microsecond=0)
    calls = db.collection("calls").where(filter=FieldFilter("owner_id", "==", user["id"])).stream()
    total_seconds = 0
    for c in calls:
        cd = c.to_dict()
        ts = cd.get("started_at", "")
        try:
            if datetime.fromisoformat(ts.replace("Z", "+00:00")) >= month_start:
                total_seconds += int(cd.get("duration_seconds", 0) or 0)
        except Exception:
            continue
    minutes_used = round(total_seconds / 60.0, 1)

    invoices_docs = db.collection("invoices").where(filter=FieldFilter("owner_id", "==", user["id"])).stream()
    invoices = [d.to_dict() for d in invoices_docs]
    invoices.sort(key=lambda x: x.get("date", ""), reverse=True)

    return _cache_set(f"billing:{user['id']}", {
        "plan": plan,
        "usage": {
            "agents": agents_count,
            "agents_limit": plan["agents_limit"],
            "minutes": minutes_used,
            "minutes_limit": plan["minutes_limit"],
            "docs": docs_count,
            "docs_limit": plan["docs_limit"],
        },
        "invoices": invoices,
    }, ttl=30)


# ----- Analytics -----

def _build_dashboard_data(uid: str) -> dict:
    """Single Firestore scan to build all dashboard data. Shared by /dashboard,
    /analytics/overview, and /analytics/agent-leaderboard.

    Performance: we only fetch the SLIM fields the dashboard needs via .select()
    — critically NOT `transcript` (which can be huge and dominated download time),
    and we cap to the most-recent N calls so the scan stays bounded as history
    grows. The agents + calls reads run in parallel threads."""
    from concurrent.futures import ThreadPoolExecutor as _TPE

    _CALL_FIELDS = ["owner_id", "agent_id", "agent_name", "started_at",
                    "duration_seconds", "status", "sentiment", "lead_tag",
                    "phone_number", "contact_name", "summary", "intent"]
    _MAX_CALLS = 1000

    def _load_calls():
        try:
            q = (db.collection("calls").where(filter=FieldFilter("owner_id", "==", uid))
                 .order_by("started_at", direction=firestore.Query.DESCENDING)
                 .limit(_MAX_CALLS))
            try:
                q = q.select(_CALL_FIELDS)
            except Exception:
                pass  # select unsupported → fall back to full docs
            return [d.to_dict() for d in q.stream()]
        except Exception:
            # Missing composite index on (owner_id, started_at) → unordered fallback.
            return [d.to_dict() for d in
                    db.collection("calls").where(filter=FieldFilter("owner_id", "==", uid)).limit(_MAX_CALLS).stream()]

    def _load_agents():
        return [d.to_dict() for d in
                db.collection("agents").where(filter=FieldFilter("owner_id", "==", uid)).stream()]

    with _TPE(max_workers=2) as ex:
        f_calls = ex.submit(_load_calls)
        f_agents = ex.submit(_load_agents)
        calls = f_calls.result()
        agents = f_agents.result()

    total_calls = len(calls)
    today = datetime.now(timezone.utc).date().isoformat()
    today_calls = [c for c in calls if c.get("started_at", "").startswith(today)]
    avg_dur = sum(c.get("duration_seconds", 0) for c in calls) / total_calls if total_calls else 0

    def count(field, val): return sum(1 for c in calls if c.get(field) == val)
    pos = count("sentiment", "positive")
    neu = count("sentiment", "neutral")
    neg = count("sentiment", "negative")
    hot = count("lead_tag", "hot")
    warm = count("lead_tag", "warm")
    cold = count("lead_tag", "cold")
    success_rate = (count("status", "completed") / total_calls * 100) if total_calls else 0

    series = []
    for i in range(13, -1, -1):
        day = (datetime.now(timezone.utc) - timedelta(days=i)).date().isoformat()
        day_calls = [c for c in calls if c.get("started_at", "").startswith(day)]
        series.append({"date": day, "calls": len(day_calls), "leads": sum(1 for c in day_calls if c.get("lead_tag") == "hot")})

    lang_counts: Dict[str, int] = {}
    for a in agents:
        lang = a.get("language", "English")
        lang_counts[lang] = lang_counts.get(lang, 0) + 1

    # Agent leaderboard from same call data
    agent_stats: Dict[str, Dict[str, Any]] = {}
    for c in calls:
        aid = c.get("agent_id")
        if not aid:
            continue
        s = agent_stats.setdefault(aid, {
            "agent_id": aid, "agent_name": c.get("agent_name", "Unknown"),
            "calls": 0, "total_duration": 0, "hot_leads": 0,
        })
        s["calls"] += 1
        s["total_duration"] += c.get("duration_seconds", 0) or 0
        if c.get("lead_tag") == "hot":
            s["hot_leads"] += 1
    leaderboard = sorted([
        {
            "agent_id": s["agent_id"], "agent_name": s["agent_name"],
            "calls": s["calls"], "avg_duration": round(s["total_duration"] / s["calls"], 1) if s["calls"] else 0,
            "hot_leads": s["hot_leads"],
        }
        for s in agent_stats.values()
    ], key=lambda x: x["calls"], reverse=True)[:10]

    # Recent calls (sorted, top 20)
    recent_calls = sorted(calls, key=lambda x: x.get("started_at", ""), reverse=True)[:20]

    overview = {
        "kpis": {
            "active_agents": sum(1 for a in agents if a.get("status") == "active"),
            "calls_today": len(today_calls),
            "calls_total": total_calls,
            "avg_duration": round(avg_dur, 1),
            "success_rate": round(success_rate, 1),
            "hot_leads": hot,
        },
        "sentiment": [{"name": "Positive", "value": pos}, {"name": "Neutral", "value": neu}, {"name": "Negative", "value": neg}],
        "leads": [{"name": "Hot", "value": hot}, {"name": "Warm", "value": warm}, {"name": "Cold", "value": cold}],
        "series": series,
        "language_usage": [{"name": k, "value": v} for k, v in lang_counts.items()],
    }
    return {"overview": overview, "leaderboard": leaderboard, "recent_calls": recent_calls}


@api_router.get("/dashboard")
def get_dashboard(user: dict = Depends(get_current_user)):
    """Single endpoint returning all dashboard data with one Firestore scan.
    Stale-while-revalidate: returns instantly from the last value while a fresh
    scan runs in the background, so the dashboard never blocks on Firestore."""
    uid = user["id"]

    def _build():
        data = _build_dashboard_data(uid)
        _cache_set(f"analytics:{uid}:overview", data["overview"], ttl=60)
        _cache_set(f"leaderboard:{uid}", data["leaderboard"], ttl=60)
        return data

    return _swr(f"dashboard:{uid}", _build, fresh_ttl=60)


@api_router.get("/analytics/overview")
def analytics_overview(user: dict = Depends(get_current_user)):
    uid = user["id"]
    cached = _cache_get(f"analytics:{uid}:overview")
    if cached is not None:
        return cached
    data = _build_dashboard_data(uid)
    _cache_set(f"leaderboard:{uid}", data["leaderboard"], ttl=60)
    _cache_set(f"dashboard:{uid}", data, ttl=60)
    return _cache_set(f"analytics:{uid}:overview", data["overview"], ttl=60)


@api_router.get("/analytics/agent-leaderboard")
def agent_leaderboard(user: dict = Depends(get_current_user)):
    uid = user["id"]
    cached = _cache_get(f"leaderboard:{uid}")
    if cached is not None:
        return cached
    data = _build_dashboard_data(uid)
    _cache_set(f"analytics:{uid}:overview", data["overview"], ttl=60)
    _cache_set(f"dashboard:{uid}", data, ttl=60)
    return _cache_set(f"leaderboard:{uid}", data["leaderboard"], ttl=60)


# ----- Lifecycle -----
@app.on_event("startup")
async def on_startup():
    logger.info("Scatter Studio API starting (Firebase + LiveKit)")


@app.on_event("shutdown")
async def shutdown_db_client():
    pass


@api_router.get("/")
async def root():
    return {"service": "Scatter Studio API", "status": "ok"}


@app.get("/")
async def app_root():
    return {"service": "Scatter Studio API", "status": "ok"}


@app.head("/")
async def app_root_head():
    return Response(status_code=200)


app.include_router(api_router)